diff --git a/docs/build-plans/01-readiness-model-wiring.md b/docs/build-plans/01-readiness-model-wiring.md index 781790e..2f27f03 100644 --- a/docs/build-plans/01-readiness-model-wiring.md +++ b/docs/build-plans/01-readiness-model-wiring.md @@ -313,6 +313,15 @@ records ──┬─> calculate_batch ────────────── ## 7. Skeleton snippets (illustrative only, do NOT paste blindly; re-verify line numbers at execution time) +> **EXECUTION CORRECTION (2026-07-07, found in adversarial review):** the sketches +> below key row-to-verdict resolution off `rec.plan_type` (the row's OWN value). +> That is WRONG when a row's plan_type differs from its dedup group's MERGED +> plan_type (conflicting groups merge to None) — it can attach another line's +> verdict, a reachable false green. The shipped implementation resolves rows +> through the dedup group key `(patient_id, device_type, DOS-as-date)` via +> `ReadinessIndex` in `core/worklist_readiness.py`. Follow the shipped code, +> not these sketches. + ### `core/worklist_readiness.py` (new) ```python diff --git a/python-backend/api/main.py b/python-backend/api/main.py index 47005c1..56bacc9 100644 --- a/python-backend/api/main.py +++ b/python-backend/api/main.py @@ -35,6 +35,13 @@ from core.persistence import ( load_doc_statuses_for_org, ) from api.normalizer import normalize_csv +from core.worklist_readiness import ( + build_readiness_index, + merged_to_record, + readiness_for_record, + readiness_stats, + resolve_group, +) app = FastAPI(title="Signal API", version="1.0.0", docs_url="/docs") @@ -166,6 +173,15 @@ class DocStateOut(BaseModel): diagnosis: str +class ReadinessItemOut(BaseModel): + doc_type: str + required_state: str # REQUIRED | NOT_REQUIRED | NOT_EVALUATED + input_state: str # SUPPLIED | ABSENT + quality: str # GOOD | PENDING | BAD | NONE + value: Optional[str] = None + satisfied: bool + + class RecordOut(BaseModel): patient_id: str device_type: str @@ -199,6 +215,11 @@ class RecordOut(BaseModel): csv_pa_status: Optional[str] = None csv_diagnosis_on_file: Optional[str] = None csv_transfer_from: Optional[str] = None + # Readiness verdict (Phase 2a coexistence — additive, ignored by the + # current frontend). readiness_status is a LineStatus value verbatim. + plan_type: Optional[str] = None + readiness_status: Optional[str] = None + readiness_items: Optional[list[ReadinessItemOut]] = None class UploadResponse(BaseModel): @@ -209,6 +230,7 @@ class UploadResponse(BaseModel): stats: dict mapping_summary: dict batch_id: str | None = None + readiness_stats: dict | None = None def _build_reason( @@ -309,7 +331,11 @@ class DocStatusRequest(BaseModel): def _to_record_out( - r, record=None, confirmed_visit_date=None, saved_doc_statuses: dict | None = None + r, + record=None, + confirmed_visit_date=None, + saved_doc_statuses: dict | None = None, + readiness=None, ) -> RecordOut: flag_val = r.flag.value if hasattr(r.flag, "value") else str(r.flag) @@ -391,6 +417,21 @@ def _to_record_out( csv_transfer_from=getattr(record, "csv_transfer_from", None) if record else None, + plan_type=getattr(record, "plan_type", None) if record else None, + readiness_status=readiness.line_status.value if readiness else None, + readiness_items=[ + ReadinessItemOut( + doc_type=i.doc_type, + required_state=i.required_state.value, + input_state=i.input_state.value, + quality=i.quality.value, + value=i.value, + satisfied=i.satisfied, + ) + for i in readiness.doc_items + ] + if readiness + else None, ) @@ -493,32 +534,52 @@ async def upload_csv( records, as_of=date.today(), confirmed_visits=confirmed_visits ) - # Build a lookup from a tuple key to original record for doc state computation + # Build a lookup from a tuple key to original record for doc state computation. record_lookup = {} for r in records: key = (r.patient_id, r.shipment_date, r.device_type, r.order_number) record_lookup[key] = r - out = [ - _to_record_out( - r, - record=record_lookup.get( - ( - r.patient_id, - r.last_shipment_date, - r.device_type, - getattr(r, "order_number", None), - ) - ), - confirmed_visit_date=confirmed_visits.get( - hashlib.sha256(r.patient_id.encode()).hexdigest() - ), - saved_doc_statuses=doc_statuses.get( - hashlib.sha256(r.patient_id.encode()).hexdigest(), {} - ), + # Readiness index (Phase 2a: additive, augments the legacy scoring — + # replaces nothing). Row-to-line membership comes from the dedup output + # itself, never re-derived from a row's own plan_type. + index = build_readiness_index(records, confirmed_visits) + + out = [] + used_lines = {} + for r in results: + verdict, line_key, merged = resolve_group( + index, r.patient_id, r.device_type, r.last_shipment_date + ) + if verdict is not None and line_key is not None: + used_lines[line_key] = verdict + # Exact row match first (byte-identical legacy path). CoverageResult + # carries no order_number, so the exact lookup misses for CSVs with an + # order-number column; the dedup merge then supplies the record fields + # (latest-non-null policy — the same values the verdict graded from). + rec = record_lookup.get( + ( + r.patient_id, + r.last_shipment_date, + r.device_type, + getattr(r, "order_number", None), + ) + ) + if rec is None and merged is not None: + rec = merged_to_record(merged) + out.append( + _to_record_out( + r, + record=rec, + confirmed_visit_date=confirmed_visits.get( + hashlib.sha256(r.patient_id.encode()).hexdigest() + ), + saved_doc_statuses=doc_statuses.get( + hashlib.sha256(r.patient_id.encode()).hexdigest(), {} + ), + readiness=verdict, + ) ) - for r in results - ] # Use real user ID and db_conn where possible if we implement it, hardcoding # IP as 0.0.0.0 for now unless req object available @@ -553,6 +614,10 @@ async def upload_csv( stats=_compute_stats(out), mapping_summary=mapping_summary, batch_id=batch_id, + # Only lines reachable from rows in THIS response, so the stats always + # reconcile with the records beside them (a record calculate_batch + # skipped cannot leave a phantom line in the counts). + readiness_stats=readiness_stats(used_lines), ) @@ -665,6 +730,9 @@ class ConfirmVisitRequest(BaseModel): csv_transfer_from: str | None = None order_number: str | None = None hcpcs: str | None = None + # Client-mapped plan type echoed from the upload — never guessed. None + # keeps the recompute honest: the verdict reads Plan Type Needed. + plan_type: str | None = None @app.post("/api/confirm-visit") @@ -747,8 +815,12 @@ async def confirm_visit( csv_transfer_from=body.csv_transfer_from, order_number=body.order_number, hcpcs=body.hcpcs, + # Same normalization the CSV path applies (lowercase + trim, never + # guessed) so RecordOut.plan_type is consistent across both paths. + plan_type=(body.plan_type or "").strip().lower() or None, ) result = calculate_coverage(record, confirmed_visit_date=confirmed) + verdict = readiness_for_record(record, confirmed_visit_date=confirmed) from core.supabase_client import get_client @@ -763,7 +835,9 @@ async def confirm_visit( db_conn=db_conn, ) - return _to_record_out(result, record=record, confirmed_visit_date=confirmed) + return _to_record_out( + result, record=record, confirmed_visit_date=confirmed, readiness=verdict + ) @app.put("/api/doc-status") diff --git a/python-backend/api/normalizer.py b/python-backend/api/normalizer.py index 1570da5..6f6b22f 100644 --- a/python-backend/api/normalizer.py +++ b/python-backend/api/normalizer.py @@ -48,6 +48,13 @@ HEADER_MAP: dict[str, list[str]] = { "plan", "plan_name", "plan name", "payer_name", "payer name", "primary_payer", "primary payer", "ins_name", "carrier", ], + # Client-mapped plan type — grading-critical, NEVER derived from the payer + # name. Bare "plan"/"plan_name" stay payer aliases (a "Plan" column in the + # wild is a payer name); stealing them would silently re-map existing CSVs. + "plan_type": [ + "plan_type", "plan type", "plan category", "plan_category", + "coverage type", "coverage_type", "insurance_type", "insurance type", + ], "component": [ "component", "item_type", "component_type", "type", "supply_type", ], @@ -312,6 +319,12 @@ def normalize_csv(text: str) -> tuple[list[ShipmentRecord], list[str], dict]: order_number = mapped.get("order_number") or None hcpcs = mapped.get("hcpcs") or None + # Client-supplied plan type, carried verbatim (lowercased and trimmed + # only — canonicalization of synonyms is a later, separate step). + # Anything the readiness engine does not recognize grades as + # "Plan Type Needed"; it is never guessed from the payer name. + plan_type = (mapped.get("plan_type") or "").strip().lower() or None + records.append(ShipmentRecord( patient_id=patient_id, device_type=device_type, @@ -327,6 +340,7 @@ def normalize_csv(text: str) -> tuple[list[ShipmentRecord], list[str], dict]: csv_transfer_from=csv_transfer_from, order_number=order_number, hcpcs=hcpcs, + plan_type=plan_type, )) return records, skipped, mapping_summary diff --git a/python-backend/core/worklist_readiness.py b/python-backend/core/worklist_readiness.py new file mode 100644 index 0000000..901c0f2 --- /dev/null +++ b/python-backend/core/worklist_readiness.py @@ -0,0 +1,203 @@ +""" +worklist_readiness.py +Signal — STTIL Solutions + +Orchestrates dedup + readiness for the live pipeline (Phase 2a coexistence, +docs/build-plans/01-readiness-model-wiring.md). The readiness verdict AUGMENTS +the coverage_calculator output: timing urgency (CoverageFlag) and documentation +completeness (LineStatus) are separate axes, combined at the worklist level. + +This is the ONLY module that imports both dedup and readiness for the API. +core/coverage_calculator.py must never import this module (dedup.py imports +FROM coverage_calculator; the reverse edge would be circular). + +LINE MEMBERSHIP IS DECIDED BY DEDUP, NEVER RE-DERIVED FROM A ROW. A row's own +plan_type can differ from its dedup group's MERGED plan_type (conflicting +groups route to None — never guessed), so resolving a row's verdict from its +own plan_type can attach the WRONG line's verdict (a reachable false green, +caught in independent review 2026-07-06). All row-level resolution goes +through the (patient_id, device_type, DOS-as-date) group key via +ReadinessIndex, which records the dedup output itself. + +Non-gradeable lines (non-CGM devices) are never graded: they carry no verdict, +matching the locked design (preserved and displayed, not graded). + +PHI CONTRACT: patient_id is the sole crosswalk key; it is hashed in any log +line (dedup._hash_pid pattern). No names, SSNs, DOBs, or contact fields. +""" + +from __future__ import annotations + +import hashlib +from dataclasses import dataclass, field +from datetime import date +from typing import Optional + +from core.coverage_calculator import ShipmentRecord +from core.dedup import ( + UNKNOWN_PLAN_TYPE, + MergedShipment, + _as_date, + assign_to_coverage_lines, + dedup, +) +from core.readiness import ReadinessVerdict, evaluate_readiness + +# (patient_id, device_type, plan_type_id) — mirrors CoverageLineKey. +LineKey = tuple[str, str, str] +# (patient_id, device_type, DOS-as-date) — the locked dedup group key. +GroupKey = tuple[str, str, date] + + +@dataclass +class ReadinessIndex: + """Dedup-derived resolution index for one upload batch. + + by_line: verdicts per gradeable coverage line + line_by_group: dedup group -> the line that group's shipment belongs to + merged_by_group: dedup group -> its MergedShipment (display-field truth) + """ + + by_line: dict[LineKey, ReadinessVerdict] = field(default_factory=dict) + line_by_group: dict[GroupKey, LineKey] = field(default_factory=dict) + merged_by_group: dict[GroupKey, MergedShipment] = field(default_factory=dict) + + +def _confirmed_for( + patient_id: str, confirmed_visits: dict[str, date] +) -> Optional[date]: + """Look up a staff-confirmed visit date by hashed patient_id. + + confirmed_visits is keyed by sha256(patient_id) hex digest, the same keying + coverage_calculator.calculate_batch uses. + """ + return confirmed_visits.get( + hashlib.sha256(patient_id.encode()).hexdigest() + ) + + +def build_readiness_index( + records: list[ShipmentRecord], + confirmed_visits: Optional[dict[str, date]] = None, +) -> ReadinessIndex: + """Dedup records into coverage lines, grade each gradeable line, and + record the group->line membership map so row-level resolution never + re-derives it. + + Grades the CURRENT shipment only (latest DOS per line — locked rule); + prior shipments are collapsed history. Non-gradeable lines are skipped. + """ + confirmed_visits = confirmed_visits or {} + merged = dedup(records) + lines = assign_to_coverage_lines(merged) + + index = ReadinessIndex() + for key, line in lines.items(): + if not line.gradeable: + continue + cur = line.current_shipment + index.by_line[(key.patient_id, key.device_type, key.plan_type_id)] = ( + evaluate_readiness( + line.plan_type if line.plan_type != UNKNOWN_PLAN_TYPE else None, + csv_swo_status=cur.csv_swo_status, + csv_visit_date=cur.csv_visit_date, + confirmed_visit_date=_confirmed_for( + key.patient_id, confirmed_visits + ), + csv_pecos_verified=cur.csv_pecos_verified, + csv_pa_status=cur.csv_pa_status, + csv_diagnosis_on_file=cur.csv_diagnosis_on_file, + ) + ) + + for m in merged: + gkey: GroupKey = (m.patient_id, m.device_type, m.shipment_date) + index.merged_by_group[gkey] = m + # Membership from the MERGED plan_type — the same rule + # assign_to_coverage_lines used, so this cannot diverge from it. + index.line_by_group[gkey] = ( + m.patient_id, + m.device_type, + m.plan_type or UNKNOWN_PLAN_TYPE, + ) + + return index + + +def readiness_by_line( + records: list[ShipmentRecord], + confirmed_visits: Optional[dict[str, date]] = None, +) -> dict[LineKey, ReadinessVerdict]: + """Line verdicts only (convenience wrapper over build_readiness_index).""" + return build_readiness_index(records, confirmed_visits).by_line + + +def resolve_group( + index: ReadinessIndex, + patient_id: str, + device_type: str, + shipment_date, +) -> tuple[Optional[ReadinessVerdict], Optional[LineKey], Optional[MergedShipment]]: + """Resolve a result row to (verdict, line_key, merged_shipment) through the + dedup group key. The only sanctioned row-level resolution path.""" + gkey: GroupKey = (patient_id, device_type, _as_date(shipment_date)) + merged = index.merged_by_group.get(gkey) + line_key = index.line_by_group.get(gkey) + verdict = index.by_line.get(line_key) if line_key else None + return verdict, line_key, merged + + +def merged_to_record(m: MergedShipment) -> ShipmentRecord: + """Synthetic ShipmentRecord from a dedup merge, used when the exact + row lookup misses (order-numbered CSVs: CoverageResult carries no + order_number). Field policy matches dedup exactly (latest-non-null doc + fields, collected order numbers), so the legacy doc_state and the + readiness verdict grade from the SAME values and cannot contradict. + order_number/hcpcs join the collected display lists — display-only fields. + """ + return ShipmentRecord( + patient_id=m.patient_id, + device_type=m.device_type, + shipment_date=m.shipment_date, + quantity=m.total_quantity, + payer=m.payer, + component=m.components[0] if m.components else "sensor", + csv_visit_date=m.csv_visit_date, + csv_swo_status=m.csv_swo_status, + csv_pecos_verified=m.csv_pecos_verified, + csv_pa_status=m.csv_pa_status, + csv_diagnosis_on_file=m.csv_diagnosis_on_file, + csv_transfer_from=m.csv_transfer_from, + order_number=", ".join(m.order_numbers) if m.order_numbers else None, + hcpcs=", ".join(m.hcpcs_codes) if m.hcpcs_codes else None, + plan_type=m.plan_type, + ) + + +def readiness_for_record( + record: ShipmentRecord, confirmed_visit_date: Optional[date] = None +) -> ReadinessVerdict: + """Grade a single record as its own line (the Confirm Visit recompute path).""" + return evaluate_readiness( + record.plan_type, + csv_swo_status=record.csv_swo_status, + csv_visit_date=record.csv_visit_date, + confirmed_visit_date=confirmed_visit_date, + csv_pecos_verified=record.csv_pecos_verified, + csv_pa_status=record.csv_pa_status, + csv_diagnosis_on_file=record.csv_diagnosis_on_file, + ) + + +def readiness_stats(verdicts: dict[LineKey, ReadinessVerdict]) -> dict: + """Line-level counts per LineStatus value, for measurable live verification. + + Callers pass only the lines actually reachable from response rows, so the + stats always reconcile with the records in the same response. + """ + counts: dict[str, int] = {} + for verdict in verdicts.values(): + label = verdict.line_status.value + counts[label] = counts.get(label, 0) + 1 + counts["lines_total"] = len(verdicts) + return counts diff --git a/tests/test_worklist_readiness.py b/tests/test_worklist_readiness.py new file mode 100644 index 0000000..14e5f3e --- /dev/null +++ b/tests/test_worklist_readiness.py @@ -0,0 +1,596 @@ +""" +Tests for the readiness-model wiring (build plan 01). + +Engine behavior is covered by test_dedup.py / test_readiness.py; these tests +target the SEAMS the wiring adds: the orchestration module, the /api/upload +augmentation, the /api/confirm-visit recompute, and back-compat invariants. + +PHI CONTRACT: fixtures use synthetic patient_ids only. No names, DOBs, SSNs. +""" + +import hashlib +import io +import sys +from datetime import date, timedelta +from pathlib import Path + +import pytest + +sys.path.insert(0, str(Path(__file__).parent.parent / "python-backend")) + +from core.coverage_calculator import ShipmentRecord, calculate_batch # noqa: E402 +from core.worklist_readiness import ( # noqa: E402 + build_readiness_index, + merged_to_record, + readiness_by_line, + readiness_for_record, + readiness_stats, + resolve_group, +) + +TODAY = date.today() + + +def _record(**kwargs) -> ShipmentRecord: + defaults = dict( + patient_id="PT-1001", + device_type="dexcom_g7", + shipment_date=TODAY - timedelta(days=5), + quantity=3, + payer="Medicare Part B", + component="sensor", + ) + defaults.update(kwargs) + return ShipmentRecord(**defaults) + + +def _hash(pid: str) -> str: + return hashlib.sha256(pid.encode()).hexdigest() + + +# --- orchestration module (unit level) -------------------------------------- + + +def test_readiness_by_line_full_evidence_is_clear_to_ship(): + rec = _record( + plan_type="medicare", + csv_swo_status="On File", + csv_visit_date=TODAY - timedelta(days=30), + csv_pecos_verified="Yes", + csv_diagnosis_on_file="Yes", + ) + verdicts = readiness_by_line([rec]) + key = ("PT-1001", "dexcom_g7", "medicare") + assert key in verdicts + assert verdicts[key].line_status.value == "Clear to Ship" + + +def test_readiness_by_line_skips_non_gradeable_lines(): + rec = _record(device_type="omnipod_5", plan_type="medicare") + verdicts = readiness_by_line([rec]) + assert verdicts == {} + + +def test_no_plan_type_reads_plan_type_needed(): + rec = _record( + csv_swo_status="On File", + csv_visit_date=TODAY - timedelta(days=30), + csv_pecos_verified="Yes", + csv_diagnosis_on_file="Yes", + ) + verdicts = readiness_by_line([rec]) + key = ("PT-1001", "dexcom_g7", "unknown") + assert verdicts[key].line_status.value == "Plan Type Needed" + by_type = {i.doc_type: i for i in verdicts[key].doc_items} + assert by_type["pecos"].required_state.value == "NOT_EVALUATED" + assert by_type["pa"].required_state.value == "NOT_EVALUATED" + + +def test_same_dos_different_order_numbers_is_one_line_verdict(): + a = _record(order_number="ORD-1", plan_type="medicare") + b = _record(order_number="ORD-2", plan_type="medicare") + verdicts = readiness_by_line([a, b]) + assert len(verdicts) == 1 + + +def test_verdict_graded_from_latest_dos(): + older = _record( + shipment_date=TODAY - timedelta(days=40), + plan_type="medicare", + csv_swo_status="On File", + csv_visit_date=TODAY - timedelta(days=45), + csv_pecos_verified="Yes", + csv_diagnosis_on_file="Yes", + ) + # Latest DOS is missing PECOS -> the line verdict must reflect THIS row. + newer = _record( + shipment_date=TODAY - timedelta(days=5), + plan_type="medicare", + csv_swo_status="On File", + csv_visit_date=TODAY - timedelta(days=45), + csv_diagnosis_on_file="Yes", + ) + verdicts = readiness_by_line([older, newer]) + assert len(verdicts) == 1 + verdict = verdicts[("PT-1001", "dexcom_g7", "medicare")] + assert verdict.line_status.value == "Action Needed" + + +def test_conflicting_plan_types_route_to_unknown_line(): + a = _record(order_number="ORD-1", plan_type="medicare") + b = _record(order_number="ORD-2", plan_type="medicaid") + index = build_readiness_index([a, b]) + assert set(index.by_line) == {("PT-1001", "dexcom_g7", "unknown")} + verdict, line_key, merged = resolve_group( + index, "PT-1001", "dexcom_g7", a.shipment_date + ) + assert verdict is not None + assert verdict.line_status.value == "Plan Type Needed" + assert line_key == ("PT-1001", "dexcom_g7", "unknown") + assert merged is not None + + +def test_mixed_plan_population_resolves_to_merged_line(): + """Review finding (HIGH family): a group where only SOME rows carry + plan_type merges to the typed line; row resolution must follow the MERGE, + not any single row's own plan_type.""" + blank = _record( + order_number="ORD-1", + csv_swo_status="On File", + csv_visit_date=TODAY - timedelta(days=30), + csv_pecos_verified="Yes", + csv_diagnosis_on_file="Yes", + ) + typed = _record( + order_number="ORD-2", + plan_type="medicare", + csv_swo_status="On File", + csv_visit_date=TODAY - timedelta(days=30), + csv_pecos_verified="Yes", + csv_diagnosis_on_file="Yes", + ) + index = build_readiness_index([blank, typed]) + assert set(index.by_line) == {("PT-1001", "dexcom_g7", "medicare")} + verdict, line_key, _ = resolve_group( + index, "PT-1001", "dexcom_g7", blank.shipment_date + ) + assert verdict is not None, "blank-plan row must inherit its merged line's verdict" + assert verdict.line_status.value == "Clear to Ship" + assert line_key == ("PT-1001", "dexcom_g7", "medicare") + + +def test_conflicted_dos_never_borrows_another_lines_verdict(): + """Review finding (HIGH): a conflicted newer DOS must read its own + unknown-line verdict, never the older clean line's green.""" + old_clean = _record( + shipment_date=TODAY - timedelta(days=45), + plan_type="medicare", + csv_swo_status="On File", + csv_visit_date=TODAY - timedelta(days=50), + csv_pecos_verified="Yes", + csv_diagnosis_on_file="Yes", + ) + new_a = _record(order_number="ORD-1", plan_type="medicare") + new_b = _record(order_number="ORD-2", plan_type="medicaid") + index = build_readiness_index([old_clean, new_a, new_b]) + assert set(index.by_line) == { + ("PT-1001", "dexcom_g7", "medicare"), + ("PT-1001", "dexcom_g7", "unknown"), + } + old_verdict, _, _ = resolve_group( + index, "PT-1001", "dexcom_g7", old_clean.shipment_date + ) + new_verdict, _, _ = resolve_group( + index, "PT-1001", "dexcom_g7", new_a.shipment_date + ) + assert old_verdict.line_status.value == "Clear to Ship" + assert new_verdict.line_status.value == "Plan Type Needed" + + +def test_merged_to_record_matches_dedup_field_policy(): + """Review finding (MEDIUM): fallback record fields must come from the + dedup merge (latest-non-null, collected order numbers), so the legacy + doc_state cannot contradict the readiness verdict.""" + first = _record(order_number="ORD-1", csv_pecos_verified="Yes") + second = _record(order_number="ORD-2", plan_type="medicare") + index = build_readiness_index([first, second]) + _, _, merged = resolve_group( + index, "PT-1001", "dexcom_g7", first.shipment_date + ) + rec = merged_to_record(merged) + assert rec.csv_pecos_verified == "Yes" # latest-non-null keeps it + assert rec.order_number == "ORD-1, ORD-2" # collected, not arbitrary + assert rec.plan_type == "medicare" + + +def test_confirmed_visit_counts_as_positive_evidence(): + rec = _record( + plan_type="medicare", + csv_swo_status="On File", + csv_pecos_verified="Yes", + csv_diagnosis_on_file="Yes", + ) + confirmed = {_hash("PT-1001"): TODAY - timedelta(days=20)} + verdicts = readiness_by_line([rec], confirmed_visits=confirmed) + verdict = verdicts[("PT-1001", "dexcom_g7", "medicare")] + assert verdict.line_status.value == "Clear to Ship" + visit = next(i for i in verdict.doc_items if i.doc_type == "visit") + assert visit.satisfied + + +def test_readiness_stats_counts_lines(): + good = _record( + patient_id="PT-A", + plan_type="medicare", + csv_swo_status="On File", + csv_visit_date=TODAY - timedelta(days=30), + csv_pecos_verified="Yes", + csv_diagnosis_on_file="Yes", + ) + unknown = _record(patient_id="PT-B") + stats = readiness_stats(readiness_by_line([good, unknown])) + assert stats["Clear to Ship"] == 1 + assert stats["Plan Type Needed"] == 1 + assert stats["lines_total"] == 2 + + +def test_no_raw_patient_id_in_logs(caplog): + # Conflicting plan types force the dedup warning path. + a = _record(patient_id="RAW-SECRET-ID-42", order_number="1", plan_type="medicare") + b = _record(patient_id="RAW-SECRET-ID-42", order_number="2", plan_type="medicaid") + with caplog.at_level("WARNING"): + readiness_by_line([a, b]) + assert caplog.text # the warning fired + assert "RAW-SECRET-ID-42" not in caplog.text + + +def test_readiness_for_record_single_path(): + rec = _record( + plan_type="medicare", + csv_swo_status="On File", + csv_pecos_verified="Yes", + csv_diagnosis_on_file="Yes", + ) + verdict = readiness_for_record( + rec, confirmed_visit_date=TODAY - timedelta(days=10) + ) + assert verdict.line_status.value == "Clear to Ship" + + +# --- API level (FastAPI TestClient) ------------------------------------------ + +httpx = pytest.importorskip("httpx", reason="httpx required for TestClient") +from fastapi.testclient import TestClient # noqa: E402 + +import api.main as api_main # noqa: E402 + +client = TestClient(api_main.app) + + +def _upload(csv_text: str): + return client.post( + "/api/upload", + files={"file": ("fixture.csv", io.BytesIO(csv_text.encode()), "text/csv")}, + ) + + +# Dates engineered so the legacy timing flag is RESUPPLY_READY (coverage ends +# within the 30-day window) and no renewal flag fires (visit recent). +SHIP = (TODAY - timedelta(days=5)).isoformat() +VISIT = (TODAY - timedelta(days=30)).isoformat() + +HEADER = ( + "patient_id,device_type,shipment_date,quantity,payer,plan_type," + "swo_status,visit_date,pecos_verified,pa_status,diagnosis_on_file" +) + + +def test_upload_false_green_blocked_at_api_layer(): + """Acceptance case 1: legacy label green-ish, readiness verdict honest.""" + csv_text = ( + HEADER + + f"\nPT-FG-1,Dexcom G7,{SHIP},3,Medicare Part B,medicare," + + f"On File,{VISIT},,,Yes" + ) + resp = _upload(csv_text) + assert resp.status_code == 200 + rec = resp.json()["records"][0] + assert rec["flag"] == "RESUPPLY_READY" + assert rec["status_label"] == "Clear to Ship" # legacy timing label + assert rec["readiness_status"] == "Action Needed" # PECOS absent blocks green + + +def test_upload_without_plan_type_reads_plan_type_needed(): + """Acceptance case 2: today's default for every record — honest, never guessed.""" + header = ( + "patient_id,device_type,shipment_date,quantity,payer," + "swo_status,visit_date,pecos_verified,pa_status,diagnosis_on_file" + ) + csv_text = ( + header + + f"\nPT-PTN-1,Dexcom G7,{SHIP},3,Medicare Part B," + + f"On File,{VISIT},Yes,,Yes" + ) + resp = _upload(csv_text) + assert resp.status_code == 200 + body = resp.json() + graded = [r for r in body["records"] if r["readiness_status"] is not None] + assert graded, "expected at least one graded record" + assert all(r["readiness_status"] == "Plan Type Needed" for r in graded) + assert body["readiness_stats"]["Plan Type Needed"] == body[ + "readiness_stats" + ]["lines_total"] + + +def test_upload_clear_to_ship_reachable_end_to_end(): + """Acceptance case 3: full positive evidence + mapped plan type = true green.""" + csv_text = ( + HEADER + + f"\nPT-GREEN-1,Dexcom G7,{SHIP},3,Medicare Part B,medicare," + + f"On File,{VISIT},Yes,,Yes" + ) + resp = _upload(csv_text) + assert resp.status_code == 200 + rec = resp.json()["records"][0] + assert rec["plan_type"] == "medicare" + assert rec["readiness_status"] == "Clear to Ship" + items = {i["doc_type"]: i for i in rec["readiness_items"]} + assert items["pa"]["required_state"] == "NOT_REQUIRED" # Medicare FFS + assert all(i["satisfied"] for i in items.values()) + + +def test_upload_with_order_numbers_still_resolves_records(): + """Regression for the record_lookup miss: CoverageResult has no + order_number, so exact-key lookups miss for order-numbered CSVs; the + fallback must recover doc_state AND readiness fields.""" + header = HEADER + ",order_number" + csv_text = ( + header + + f"\nPT-ORD-1,Dexcom G7,{SHIP},3,Medicare Part B,medicare," + + f"On File,{VISIT},Yes,,Yes,ORD-778899" + ) + resp = _upload(csv_text) + assert resp.status_code == 200 + rec = resp.json()["records"][0] + assert rec["order_number"] == "ORD-778899" + assert rec["doc_state"] is not None + assert rec["readiness_status"] == "Clear to Ship" + + +def test_upload_backcompat_legacy_fields_unchanged(): + """Acceptance case 6: the wiring is additive — every legacy field matches + an independent reconstruction from the engines for an order-less CSV + (where the exact record lookup must hit and byte-identity must hold).""" + csv_text = ( + HEADER + + f"\nPT-BC-1,Dexcom G7,{SHIP},3,Medicare Part B,medicare," + + f"On File,{VISIT},Yes,,Yes" + + f"\nPT-BC-2,Libre 3,{(TODAY - timedelta(days=90)).isoformat()},1," + + "Aetna,commercial,Pending,,,Requested,Yes" + ) + resp = _upload(csv_text) + assert resp.status_code == 200 + body = resp.json() + + from api.normalizer import normalize_csv + from core.doc_state_machine import compute_doc_state + + records, _, _ = normalize_csv(csv_text) + by_pid = {r.patient_id: r for r in records} + expected = calculate_batch(records, as_of=TODAY, confirmed_visits={}) + + assert len(body["records"]) == len(expected) + for got, want in zip(body["records"], expected): + rec = by_pid[want.patient_id] + assert got["patient_id"] == want.patient_id + assert got["flag"] == want.flag.value + assert got["priority_score"] == want.priority_score + assert got["status_label"] == api_main.FLAG_LABELS.get( + want.flag.value, want.flag.value + ) + assert got["payer"] == want.payer + assert got["days_until_coverage_end"] == want.days_until_coverage_end + assert got["coverage_end_date"] == want.coverage_end_date.isoformat() + assert got["rule_version"] == want.rule_version + assert got["action"] == api_main.FLAG_ACTIONS.get( + want.flag.value, "Review" + ) + # Record-echo fields resolve through the exact lookup (order-less CSV) + assert got["quantity"] == rec.quantity + assert got["order_number"] == rec.order_number + assert got["csv_swo_status"] == rec.csv_swo_status + assert got["csv_pecos_verified"] == rec.csv_pecos_verified + assert got["csv_pa_status"] == rec.csv_pa_status + assert got["csv_diagnosis_on_file"] == rec.csv_diagnosis_on_file + # doc_state reconstructed independently from the same inputs + doc = compute_doc_state( + payer_type=api_main._normalize_payer_type(want.payer), + csv_swo_status=rec.csv_swo_status, + csv_visit_date=rec.csv_visit_date, + confirmed_visit_date=None, + csv_pecos_verified=rec.csv_pecos_verified, + csv_pa_status=rec.csv_pa_status, + csv_diagnosis_on_file=rec.csv_diagnosis_on_file, + is_transfer=bool(rec.csv_transfer_from), + ) + assert got["doc_state"]["visit"] == doc.visit + assert got["doc_state"]["pecos"] == doc.pecos + assert got["doc_state"]["diagnosis"] == doc.diagnosis + assert got["cascade"] == doc.cascade + + # Full legacy stats dict reconstructed independently from the flags + flags = [w.flag.value for w in expected] + assert body["stats"] == { + "total": len(expected), + "supply_lapsed": flags.count("SUPPLY_LAPSED"), + "visit_required": flags.count("VISIT_REQUIRED"), + "transfer_pending": flags.count("TRANSFER_PENDING"), + "renewal_critical": flags.count("RENEWAL_CRITICAL"), + "renewal_elevated": flags.count("RENEWAL_ELEVATED"), + "renewal_soon": flags.count("RENEWAL_SOON"), + "resupply_ready": flags.count("RESUPPLY_READY"), + "active": flags.count("ACTIVE"), + "no_recent_shipment": flags.count("NO_RECENT_SHIPMENT"), + "prescriber_action": ( + flags.count("SUPPLY_LAPSED") + + flags.count("VISIT_REQUIRED") + + flags.count("RENEWAL_CRITICAL") + + flags.count("RENEWAL_ELEVATED") + ), + } + + +def test_upload_mixed_plan_group_both_rows_carry_merged_verdict(): + """Review finding (HIGH family), API level: duplicate order-numbered rows + where only one carries plan_type — both output rows must carry the merged + line's verdict, and stats must reconcile with the rows.""" + header = HEADER + ",order_number" + csv_text = ( + header + + f"\nPT-MIX-1,Dexcom G7,{SHIP},3,Medicare Part B,," + + f"On File,{VISIT},Yes,,Yes,ORD-1" + + f"\nPT-MIX-1,Dexcom G7,{SHIP},3,Medicare Part B,medicare," + + f"On File,{VISIT},Yes,,Yes,ORD-2" + ) + resp = _upload(csv_text) + assert resp.status_code == 200 + body = resp.json() + assert len(body["records"]) == 2 + for rec in body["records"]: + assert rec["readiness_status"] == "Clear to Ship" + assert rec["plan_type"] == "medicare" + assert rec["order_number"] == "ORD-1, ORD-2" + assert body["readiness_stats"] == {"Clear to Ship": 1, "lines_total": 1} + + +def test_upload_conflicted_dos_reads_own_line_not_older_green(): + """Review finding (HIGH), API level: the false-green repro. A conflicted + newer DOS must read Plan Type Needed even when an older clean DOS on the + same patient+device graded Clear to Ship.""" + header = HEADER + ",order_number" + old_ship = (TODAY - timedelta(days=45)).isoformat() + old_visit = (TODAY - timedelta(days=50)).isoformat() + csv_text = ( + header + + f"\nPT-CONF-1,Dexcom G7,{old_ship},3,Medicare Part B,medicare," + + f"On File,{old_visit},Yes,,Yes,ORD-OLD" + + f"\nPT-CONF-1,Dexcom G7,{SHIP},3,Medicare Part B,medicare," + + f"On File,{VISIT},Yes,,Yes,ORD-N1" + + f"\nPT-CONF-1,Dexcom G7,{SHIP},3,Medicaid - GA,medicaid," + + f"On File,{VISIT},Yes,,Yes,ORD-N2" + ) + resp = _upload(csv_text) + assert resp.status_code == 200 + body = resp.json() + by_ship = {} + for rec in body["records"]: + by_ship.setdefault(rec["last_shipment_date"], []).append(rec) + for rec in by_ship[old_ship]: + assert rec["readiness_status"] == "Clear to Ship" + for rec in by_ship[SHIP]: + assert rec["readiness_status"] == "Plan Type Needed" + assert body["readiness_stats"] == { + "Clear to Ship": 1, + "Plan Type Needed": 1, + "lines_total": 2, + } + + +def test_upload_skipped_record_leaves_no_phantom_line_in_stats(): + """Review finding (LOW): a record calculate_batch skips (G7 transmitter + has no wear-day rule) must not leave a phantom line in readiness_stats.""" + header = HEADER + ",component" + csv_text = ( + header + + f"\nPT-OK-1,Dexcom G7,{SHIP},3,Medicare Part B,medicare," + + f"On File,{VISIT},Yes,,Yes,sensor" + + f"\nPT-TX-1,Dexcom G7,{SHIP},1,Medicare Part B,medicare," + + f"On File,{VISIT},Yes,,Yes,transmitter" + ) + resp = _upload(csv_text) + assert resp.status_code == 200 + body = resp.json() + pids = {r["patient_id"] for r in body["records"]} + assert pids == {"PT-OK-1"} # transmitter row skipped by the engine + assert body["readiness_stats"]["lines_total"] == 1 + + +def test_upload_non_gradeable_rows_carry_no_verdict(): + csv_text = ( + HEADER + + f"\nPT-POD-1,Omnipod 5,{SHIP},1,Medicare Part B,medicare," + + f"On File,{VISIT},Yes,,Yes" + ) + resp = _upload(csv_text) + assert resp.status_code == 200 + rec = resp.json()["records"][0] + assert rec["readiness_status"] is None + assert rec["readiness_items"] is None + + +def test_confirm_visit_recompute_flips_to_clear(monkeypatch): + """Acceptance case 5: confirming the visit satisfies the last open + required item in the same response.""" + monkeypatch.setattr(api_main, "get_or_create_org", lambda **kw: "org-test") + monkeypatch.setattr( + api_main, "upsert_confirmed_visit", lambda *a, **kw: True + ) + + # Upload path first: everything good except visit evidence. + csv_text = ( + HEADER + + f"\nPT-CV-1,Dexcom G7,{SHIP},3,Medicare Part B,medicare," + + "On File,,Yes,,Yes" + ) + up = _upload(csv_text) + assert up.status_code == 200 + assert up.json()["records"][0]["readiness_status"] == "Action Needed" + + resp = client.post( + "/api/confirm-visit", + json={ + "patient_id": "PT-CV-1", + "confirmed_date": (TODAY - timedelta(days=15)).isoformat(), + "shipment_date": SHIP, + "payer": "Medicare Part B", + "device_type": "dexcom_g7", + "quantity": 3, + "csv_swo_status": "On File", + "csv_pecos_verified": "Yes", + "csv_diagnosis_on_file": "Yes", + "plan_type": "medicare", + }, + ) + assert resp.status_code == 200 + rec = resp.json() + assert rec["readiness_status"] == "Clear to Ship" + visit = next(i for i in rec["readiness_items"] if i["doc_type"] == "visit") + assert visit["satisfied"] + + +def test_confirm_visit_normalizes_plan_type(monkeypatch): + """Review finding (LOW): both paths must return the same normalized + plan_type representation.""" + monkeypatch.setattr(api_main, "get_or_create_org", lambda **kw: "org-test") + monkeypatch.setattr( + api_main, "upsert_confirmed_visit", lambda *a, **kw: True + ) + resp = client.post( + "/api/confirm-visit", + json={ + "patient_id": "PT-NORM-1", + "confirmed_date": (TODAY - timedelta(days=15)).isoformat(), + "shipment_date": SHIP, + "payer": "Medicare Part B", + "device_type": "dexcom_g7", + "quantity": 3, + "csv_swo_status": "On File", + "csv_pecos_verified": "Yes", + "csv_diagnosis_on_file": "Yes", + "plan_type": " Medicare ", + }, + ) + assert resp.status_code == 200 + rec = resp.json() + assert rec["plan_type"] == "medicare" + assert rec["readiness_status"] == "Clear to Ship"