Compare commits
No commits in common. "4517b0b89fe9c503f3ae188e9ee93aaa33cd7c3d" and "e3de616023812ba96371ec924b6cb769cbe6d32d" have entirely different histories.
4517b0b89f
...
e3de616023
7 changed files with 5 additions and 1154 deletions
|
|
@ -2,16 +2,12 @@ v:1
|
|||
|
||||
# current-state
|
||||
|
||||
ACTIVE: 2026-06-26 Pi session. STTIL AIOS architecture map DESIGNED and WRITTEN (STTIL-Vault/Projects/STTIL-AIOS-architecture-map.md, 338 lines, 10 sections). Claude SHIPPED readiness-model core engines (dedup + verdict) — additive only, not yet wired. Pi's AIOS map confirms Layers 1-3 are solid; readiness model is unblocked. See NEXT for remaining build steps.
|
||||
ACTIVE: 2026-06-25 was a non-Signal session. labABLE CRUSH brief + verified, 3-model-cross-checked research staged for Kisa to send; Perplexity MCP fixed (BASE_URL path bug); OpenRouter `ask` multi-model CLI wired (~/scripts/ask.py); Pi briefed to design the STTIL AIOS map (Projects/pi-aios-design-brief-2026-06-25.md, write the full file in sections). Signal readiness-model build is unchanged and remains the top Signal priority (dedup LOCKED, see NEXT).
|
||||
|
||||
NEXT (in order): (1) Kisa: send the labABLE CRUSH brief + schedule follow-up call (staged at ~/Documents/labABLE/). (2) Claude: continue readiness-model build — rules-to-addressable-config (citation IDs), patient rollup, device-keyed override + recompute + rollup, require-plan-type enforcement + remove name-guesser/default in LIVE path. (3) Pi: design frontend nesting + "plan type needed" UX label mapping (Claude flagged this as a Pi task in readiness.py). (4) Claude: create signal-convex repo + build Phase 1 of Convex-Signal spike. (5) Claude: research-versioning Layer-1 + add signal/docs/ to scope.
|
||||
|
||||
[AIOS build status] The map is written. Highest-impact build is the readiness model, NOT AIOS enhancements. AIOS Phase 1 items (reasoning frameworks, Playground, /board skill) are <5 hours total and optional until Signal has customer momentum.
|
||||
NEXT (in order): (1) Claude: build the readiness model — dedup decision is LOCKED (docs/dedup-design-2026-06-24.md). Build sequence: rename coverage->readiness -> extract rules to addressable config (IDs) -> patient/line/shipment grouping + required-vs-input verdict -> device-keyed override + recompute + rollup -> require-plan-type + client value-mapping + remove guesser/default -> frontend nesting + "plan type needed" state. (2) Pi: design the STTIL AIOS architecture map from Projects/pi-aios-design-brief-2026-06-25.md (write the full file). (3) Kisa: send the labABLE CRUSH brief + schedule the follow-up call. (4) Claude: create signal-convex repo (GitHub via `gh`, Forgejo needs Kisa's token) + build Phase 1 of the Convex-Signal spike from Pi's design (docs/convex-signal-spike-design-2026-06-24.md). (5) Claude: research-versioning Layer-1 + add signal/docs/ to its scope (Audit Finding 5).
|
||||
|
||||
## Decisions (last 8)
|
||||
|
||||
- 2026-06-26 [Pi]: STTIL AIOS architecture map DESIGNED and WRITTEN to STTIL-Vault/Projects/STTIL-AIOS-architecture-map.md (338 lines, 10 sections). 6-layer architecture with per-layer Have/Partial/Gap status, build sequence in 3 phases, and comparison to coach's framework. Kisa is ahead of coach's L2 AIOS in memory breadth, multi-model research, and cross-session state protocol. Highest-impact build remains the Signal readiness model.
|
||||
- 2026-06-26 [CC]: Readiness-model core engines SHIPPED — dedup.py (250 lines, 30 tests) + readiness.py (286 lines, 27 tests). All 57 tests green. Dedup key = (patient_id, device_type, DOS). Verdict engine kills the false green (required+absent blocks green, plan unknown = Plan Type Needed). Additive only, not yet wired into live pipeline. Remaining: rules-to-config, patient rollup, device override, plan-type enforcement, frontend nesting.
|
||||
- 2026-06-25 [CC]: labABLE call done (CRUSH/audit-readiness angle). CRUSH brief + verified, 3-model-cross-checked research (Opus agents, Sonnet 4.6, Perplexity Sonar) staged for Kisa to send (Projects/labABLE-CRUSH-brief-2026-06-25.md). Perplexity MCP fixed (BASE_URL=/chat/completions). OpenRouter `ask` multi-model CLI wired. Pi briefed to design the STTIL AIOS map. Display corruption diagnosed as macOS 26.0.1 GPU issue (fix = update to 26.5.1).
|
||||
- 2026-06-24 [Pi/Kisa]: Dedup decision LOCKED — key = (patient_id, device_type, date_of_service). Order_number is display-only, never in dedup key. Scales to all DMEPOS categories. Dedup design (tiebreakers, quantity merge guard, test cases) at docs/dedup-design-2026-06-24.md. Readiness-model build is UNBLOCKED.
|
||||
- 2026-06-24 [Pi]: Convex-Signal spike full design persisted to docs/convex-signal-spike-design-2026-06-24.md. Module map with 6 swappable seams (auth, data, readiness, storage, automations, notifications), Repository interface for backend portability, Convex schema (8 tables), Convex-first auth (Clerk swappable), phased 40h/5-day plan. Claude builds Phase 1.
|
||||
|
|
@ -27,8 +23,7 @@ NEXT (in order): (1) Kisa: send the labABLE CRUSH brief + schedule follow-up cal
|
|||
- DATA-PLATFORM DECISION (2026-06-24): NocoDB leads broke (no-PK, corrupted source). Options: repaired-NocoDB / Supabase / Neon / AWS / Convex. Leads = non-PHI. Informed by the Convex spike. Interim: capture works; 2 test rows un-deletable.
|
||||
- READINESS MODEL BUILD: UNBLOCKED. Dedup decision locked (docs/dedup-design-2026-06-24.md). Ready for Claude to build (top Signal priority).
|
||||
- labABLE: call DONE 2026-06-25 (CRUSH/audit-readiness). Kisa to send the staged CRUSH brief (Projects/labABLE-CRUSH-brief-2026-06-25.md) + schedule the follow-up call. Internal-only prep at Projects/labABLE-next-call-prep-INTERNAL-2026-06-25.md.
|
||||
- Pi: design the STTIL AIOS architecture map — DONE (STTIL-Vault/Projects/STTIL-AIOS-architecture-map.md, 338 lines, file written in full). Pi-side frontend nesting + "plan type needed" UX design flagged by Claude as next Pi task.
|
||||
- READINESS MODEL BUILD: core engines shipped (dedup + verdict, 57 tests). Remaining: rules-to-config citation IDs, patient rollup, device override, plan-type enforcement, frontend nesting. Not yet wired into live pipeline — additive-only.
|
||||
- Pi: design the STTIL AIOS architecture map (brief: Projects/pi-aios-design-brief-2026-06-25.md). Pi has been truncating — brief says write the file in sections/append, not a chat summary.
|
||||
|
||||
## Stack
|
||||
|
||||
|
|
@ -45,4 +40,4 @@ NEXT (in order): (1) Kisa: send the labABLE CRUSH brief + schedule follow-up cal
|
|||
|
||||
State protocol: read = `git pull --rebase` then read this file (warn if Updated > 24h old); write = edit ACTIVE/NEXT, append one author-stamped decision, then commit + push. Full protocol in the `state-sync` skill. Detailed ship-status (what shipped / didn't / why) = docs/ship-status-ledger.md. TriLane = searchable archive, not live state.
|
||||
|
||||
## Updated: 2026-06-26 (Pi: STTIL AIOS architecture map designed + written to file. CC: readiness-model core engines shipped (dedup + verdict, 57 tests green, additive not wired). State updated to reflect both milestones.)
|
||||
## Updated: 2026-06-25 (CC: non-Signal session wrap. labABLE CRUSH brief + 3-model-cross-checked research staged to send; Perplexity MCP fixed; OpenRouter ask CLI wired; Pi briefed for the STTIL AIOS map; display corruption = macOS 26.0.1 (fix 26.5.1). Signal readiness build unchanged.)
|
||||
|
|
|
|||
|
|
@ -18,7 +18,7 @@ Canonical, git-tracked record of **what shipped, what didn't, and the reason or
|
|||
| 6 | Convex-Signal spike **design** | Pi | **SHIPPED** | docs/convex-signal-spike-design-2026-06-24.md. 40h/3-phase plan; Module map + Repository interface + Convex schema + Auth seam (Convex default, Clerk swappable) + Phase 1 (core readiness loop) = what Claude builds first. | Claude builds Phase 1 from it |
|
||||
| 7 | `signal-convex` repo creation | Claude (GitHub) + Kisa/token (Forgejo) | **PENDING** | Dependency: Pi design + Kisa approval (create once, correctly). Claude can do GitHub via `gh`; Forgejo has no API token here. | after design approved |
|
||||
| 8 | Convex BAA email (plan/cost/signing) | Kisa | **IN PROGRESS** | Kisa handling | Kisa sends |
|
||||
| 9 | Readiness-model build (rename coverage→readiness, rules-to-config, grouping+verdict, device override, require-plan-type, frontend nesting) | Claude | **IN PROGRESS** | Step 3 SHIPPED 2026-06-26/27 (additive, not yet wired): (a) `core/dedup.py` grouping (CSVRow→MergedShipment→CoverageLine), 30 tests, auditor-reviewed; (b) `core/readiness.py` verdict engine — required-vs-input two-axis grading that KILLS the false green (required+absent never satisfied; clear-to-ship only on positive evidence; plan unknown→"Plan Type Needed"; required-ness read from payer_rules.json not Python sets), 27 tests, auditor-reviewed (H1/H2 config-surface false greens + M2/M3/M4 fixed). Added additive `plan_type` to ShipmentRecord. **OPEN for Kisa/Pi:** provisional gap→label mapping (At Risk/Action Needed/On Track) + H3 scope boundary (visit recency = timing layer, not readiness verdict — confirm). Remaining: rules-to-addressable-config (citation IDs), patient rollup, device-keyed override+recompute+rollup, require-plan-type enforcement + remove name-guesser/default in the LIVE path, frontend nesting + "plan type needed" UX (→ Pi). | Next: hand frontend/label UX to Pi (brief staged); build patient rollup + WIRE orchestration in `api/main.py`/new pipeline module, NOT inside `coverage_calculator` (circular import). Each step auditor-reviewed. |
|
||||
| 9 | Readiness-model build (rename coverage→readiness, rules-to-config, grouping+verdict, device override, require-plan-type, frontend nesting) | Claude | **UNBLOCKED** | Dedup decision locked: (patient_id, device_type, DOS). Order_number is display-only. Scales to all DMEPOS categories. Design: `docs/dedup-design-2026-06-24.md`. | Claude builds from dedup design + readiness model brief |
|
||||
| 10 | NocoDB leads / CRM cleanup | Claude + Kisa | **BLOCKED** | Reason: leads table has no primary key (Signal-base data source `bdmhi02nks1pg6g` is corrupted — a fresh table there also came up PK-less; content base is healthy). SQLite can't add a PK to a populated table (Route 1 failed). Interim: capture still works (inserts via n8n `signal-cgm-lead`); the 2 test rows can't be deleted. | resolved by the data-platform decision (item 11) |
|
||||
| 11 | Data-platform decision (leads + future PHI backend) | Kisa (with Claude/Pi analysis) | **OPEN** | Options analyzed: repaired-NocoDB / Supabase Postgres / Neon / AWS (already set up, BAA+RDS+S3 parked) / Convex ($25/mo incl. BAA). Leads = non-PHI (stay outside PHI boundary). Decision informed by the Convex spike. | run spike → decide |
|
||||
| 12 | Research-versioning Layer-1 (`research-autopush.sh` + Stop-hook entry) | Claude | **NOT STARTED** | Dependency: Pi design exists (`docs/research-versioning-design-2026-06-24.md`). Lower priority than product. | next infra slot |
|
||||
|
|
|
|||
|
|
@ -76,10 +76,6 @@ class ShipmentRecord:
|
|||
csv_transfer_from: Optional[str] = None # prior supplier name if transfer
|
||||
order_number: Optional[str] = None # pass-through — billing system order/claim number
|
||||
hcpcs: Optional[str] = None # pass-through — HCPCS code from supplier CSV
|
||||
# Client-mapped plan type — REQUIRED for grading, NEVER derived from the payer name.
|
||||
# medicare | medicare_advantage | medicaid | commercial | None(=plan type needed).
|
||||
# The `payer` field above is the raw display name only (locked 2026-06-23).
|
||||
plan_type: Optional[str] = None
|
||||
|
||||
|
||||
@dataclass
|
||||
|
|
|
|||
|
|
@ -1,250 +0,0 @@
|
|||
"""
|
||||
dedup.py
|
||||
Signal — STTIL Solutions
|
||||
|
||||
Groups normalized shipment rows into deduped shipments and gradeable coverage
|
||||
lines. This is the structural foundation of the readiness model: it turns a
|
||||
flat list of CSV-derived rows into the patient -> device -> shipment hierarchy
|
||||
the worklist needs, BEFORE any grading happens.
|
||||
|
||||
Design: docs/dedup-design-2026-06-24.md (Pi, locked by Kisa 2026-06-24)
|
||||
docs/readiness-model-brief-2026-06-23.md sections 3, 5, 7
|
||||
|
||||
LOCKED DEDUP KEY: (patient_id, device_type, date_of_service)
|
||||
order_number is a displayed attribute, NEVER part of the dedup key.
|
||||
Monthly resupplies are separate dates of service, NOT duplicates.
|
||||
|
||||
PLAN TYPE: carried through verbatim from the client-mapped field. It is NEVER
|
||||
derived from the payer name. None means "plan type needed" -> keyed as
|
||||
'unknown' so the line still forms but cannot be graded green later.
|
||||
|
||||
PHI CONTRACT: operates only on patient_id, device_type, shipment_date,
|
||||
quantity, payer (display), plan_type, component, order_number, hcpcs, and
|
||||
the csv_* doc-status fields. No names, SSNs, DOBs, or contact fields.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import logging
|
||||
from collections import defaultdict
|
||||
from dataclasses import dataclass, field
|
||||
from datetime import date, datetime
|
||||
from typing import Optional
|
||||
|
||||
from core.coverage_calculator import ShipmentRecord, CoverageFlag
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def _hash_pid(patient_id: str) -> str:
|
||||
"""Short hashed patient_id for log lines — raw patient_id is never logged."""
|
||||
return hashlib.sha256(patient_id.encode()).hexdigest()[:12]
|
||||
|
||||
|
||||
def _as_date(value) -> date:
|
||||
"""Coerce a datetime to a pure date so same-day timestamps group together."""
|
||||
return value.date() if isinstance(value, datetime) else value
|
||||
|
||||
# CGM device types graded this phase. Pumps and other DMEPOS categories are
|
||||
# preserved and displayed but not graded (gradeable=False). Mirrors the CGM
|
||||
# devices in payer_rules.json; moves into addressable config in a later step.
|
||||
CGM_DEVICE_TYPES = frozenset(
|
||||
{"dexcom_g6", "dexcom_g7", "freestyle_libre_2", "freestyle_libre_3"}
|
||||
)
|
||||
|
||||
UNKNOWN_PLAN_TYPE = "unknown"
|
||||
|
||||
|
||||
def is_cgm_device(device_type: str) -> bool:
|
||||
"""True only for CGM device types graded in the current phase."""
|
||||
return device_type in CGM_DEVICE_TYPES
|
||||
|
||||
|
||||
@dataclass
|
||||
class MergedShipment:
|
||||
"""One deduped shipment: all rows sharing (patient_id, device_type, DOS)."""
|
||||
patient_id: str
|
||||
device_type: str
|
||||
shipment_date: date
|
||||
total_quantity: int
|
||||
components: list[str] # combined, deduped, first-seen order
|
||||
order_numbers: list[str] # collected, deduped — display only
|
||||
hcpcs_codes: list[str] # collected, deduped — display only
|
||||
payer: str # raw display NAME, latest non-null
|
||||
plan_type: Optional[str] # client-mapped; NEVER from payer name
|
||||
# csv_* doc fields — latest non-null wins across the group
|
||||
csv_visit_date: Optional[date] = None
|
||||
csv_swo_status: Optional[str] = None
|
||||
csv_pecos_verified: Optional[str] = None
|
||||
csv_pa_status: Optional[str] = None
|
||||
csv_diagnosis_on_file: Optional[str] = None
|
||||
csv_transfer_from: Optional[str] = None
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class CoverageLineKey:
|
||||
patient_id: str
|
||||
device_type: str
|
||||
plan_type_id: str
|
||||
|
||||
|
||||
@dataclass
|
||||
class CoverageLine:
|
||||
"""A gradeable unit: one patient's one device under one plan type.
|
||||
|
||||
line_status and timing_flag are computed later by the grading step
|
||||
(gradeLine); a freshly assigned line is intentionally ungraded.
|
||||
"""
|
||||
patient_id: str
|
||||
device_type: str
|
||||
plan_type: str
|
||||
gradeable: bool
|
||||
current_shipment: MergedShipment
|
||||
prior_shipments: list[MergedShipment] = field(default_factory=list)
|
||||
line_status: Optional[str] = None
|
||||
timing_flag: Optional[CoverageFlag] = None
|
||||
|
||||
|
||||
def _qty(record: ShipmentRecord) -> int:
|
||||
"""A row without a quantity is still a shipment -> default to 1."""
|
||||
return 1 if record.quantity is None else record.quantity
|
||||
|
||||
|
||||
def _merge_quantities(rows: list[ShipmentRecord]) -> int:
|
||||
"""Sum quantities across DISTINCT component configs, max within a config.
|
||||
|
||||
Within one component config, identical rows are likely data-entry
|
||||
duplicates, so take the MAX (guards against doubling). Across distinct
|
||||
configs (a real component split, e.g. sensor + transmitter), SUM.
|
||||
|
||||
A disagreement within one config (e.g. qty 3 and qty 100 for the same
|
||||
component) collapses to the max and leaves a warning, so silently dropping
|
||||
quantity always leaves an audit breadcrumb.
|
||||
"""
|
||||
by_config: dict[frozenset, list[int]] = defaultdict(list)
|
||||
for row in rows:
|
||||
by_config[frozenset({row.component})].append(_qty(row))
|
||||
|
||||
total = 0
|
||||
for config, qtys in by_config.items():
|
||||
if len(set(qtys)) > 1:
|
||||
logger.warning(
|
||||
"Quantity disagreement in dedup group patient=%s device=%s "
|
||||
"component=%s qtys=%s -> taking max",
|
||||
_hash_pid(rows[0].patient_id), rows[0].device_type,
|
||||
set(config), qtys,
|
||||
)
|
||||
total += max(qtys)
|
||||
return total
|
||||
|
||||
|
||||
def _merge_plan_type(rows: list[ShipmentRecord]) -> Optional[str]:
|
||||
"""Plan type is grading-critical and must NEVER be guessed.
|
||||
|
||||
If a dedup group carries more than one distinct non-null plan type, the data
|
||||
is ambiguous, so return None ('plan type needed') rather than pick one.
|
||||
"""
|
||||
distinct = {r.plan_type for r in rows if r.plan_type}
|
||||
if len(distinct) > 1:
|
||||
logger.warning(
|
||||
"Conflicting plan_type in dedup group patient=%s device=%s: %s "
|
||||
"-> routing to 'plan type needed'",
|
||||
_hash_pid(rows[0].patient_id), rows[0].device_type, sorted(distinct),
|
||||
)
|
||||
return None
|
||||
return next(iter(distinct)) if distinct else None
|
||||
|
||||
|
||||
def _dedup_preserve_order(values: list[Optional[str]]) -> list[str]:
|
||||
"""Collect non-empty values, dedup, preserve first-seen order."""
|
||||
seen: dict[str, None] = {}
|
||||
for v in values:
|
||||
if v:
|
||||
seen.setdefault(v, None)
|
||||
return list(seen.keys())
|
||||
|
||||
|
||||
def _latest_non_null(rows: list[ShipmentRecord], attr: str):
|
||||
"""Last meaningful value for an attribute, in ingestion order.
|
||||
|
||||
Empty / whitespace-only strings are treated as null so a blank later row
|
||||
never clobbers an earlier real value (does not rely on upstream nulling).
|
||||
"""
|
||||
result = None
|
||||
for row in rows:
|
||||
value = getattr(row, attr)
|
||||
if value is None:
|
||||
continue
|
||||
if isinstance(value, str) and value.strip() == "":
|
||||
continue
|
||||
result = value
|
||||
return result
|
||||
|
||||
|
||||
def dedup(records: list[ShipmentRecord]) -> list[MergedShipment]:
|
||||
"""Group rows by (patient_id, device_type, shipment_date) into MergedShipments."""
|
||||
groups: dict[tuple, list[ShipmentRecord]] = {}
|
||||
for record in records:
|
||||
key = (record.patient_id, record.device_type, _as_date(record.shipment_date))
|
||||
groups.setdefault(key, []).append(record)
|
||||
|
||||
merged: list[MergedShipment] = []
|
||||
for (patient_id, device_type, shipment_date), rows in groups.items():
|
||||
merged.append(
|
||||
MergedShipment(
|
||||
patient_id=patient_id,
|
||||
device_type=device_type,
|
||||
shipment_date=shipment_date,
|
||||
total_quantity=_merge_quantities(rows),
|
||||
components=_dedup_preserve_order([r.component for r in rows]),
|
||||
order_numbers=_dedup_preserve_order([r.order_number for r in rows]),
|
||||
hcpcs_codes=_dedup_preserve_order([r.hcpcs for r in rows]),
|
||||
payer=_latest_non_null(rows, "payer") or "",
|
||||
plan_type=_merge_plan_type(rows),
|
||||
csv_visit_date=_latest_non_null(rows, "csv_visit_date"),
|
||||
csv_swo_status=_latest_non_null(rows, "csv_swo_status"),
|
||||
csv_pecos_verified=_latest_non_null(rows, "csv_pecos_verified"),
|
||||
csv_pa_status=_latest_non_null(rows, "csv_pa_status"),
|
||||
csv_diagnosis_on_file=_latest_non_null(rows, "csv_diagnosis_on_file"),
|
||||
csv_transfer_from=_latest_non_null(rows, "csv_transfer_from"),
|
||||
)
|
||||
)
|
||||
|
||||
merged.sort(key=lambda m: (m.patient_id, m.device_type, m.shipment_date))
|
||||
return merged
|
||||
|
||||
|
||||
def assign_to_coverage_lines(
|
||||
merged_shipments: list[MergedShipment],
|
||||
) -> dict[CoverageLineKey, CoverageLine]:
|
||||
"""Group MergedShipments into CoverageLines by (patient_id, device_type, plan_type).
|
||||
|
||||
plan_type is the client-mapped field. None is keyed as 'unknown' (plan type
|
||||
needed) so the line still forms but cannot be graded green later. The latest
|
||||
DOS in each line is the current shipment; earlier ones are collapsed history.
|
||||
"""
|
||||
by_key: dict[CoverageLineKey, list[MergedShipment]] = {}
|
||||
for shipment in merged_shipments:
|
||||
key = CoverageLineKey(
|
||||
patient_id=shipment.patient_id,
|
||||
device_type=shipment.device_type,
|
||||
plan_type_id=shipment.plan_type or UNKNOWN_PLAN_TYPE,
|
||||
)
|
||||
by_key.setdefault(key, []).append(shipment)
|
||||
|
||||
lines: dict[CoverageLineKey, CoverageLine] = {}
|
||||
for key, shipments in by_key.items():
|
||||
shipments.sort(key=lambda s: s.shipment_date, reverse=True)
|
||||
lines[key] = CoverageLine(
|
||||
patient_id=key.patient_id,
|
||||
device_type=key.device_type,
|
||||
plan_type=key.plan_type_id,
|
||||
gradeable=is_cgm_device(key.device_type),
|
||||
current_shipment=shipments[0],
|
||||
prior_shipments=shipments[1:],
|
||||
line_status=None,
|
||||
timing_flag=None,
|
||||
)
|
||||
|
||||
return lines
|
||||
|
|
@ -1,286 +0,0 @@
|
|||
"""
|
||||
readiness.py
|
||||
Signal — STTIL Solutions
|
||||
|
||||
The readiness verdict engine: grades ONE coverage line's documentation against
|
||||
the five tracked items, separating two axes the old code conflated —
|
||||
|
||||
required_state: does this plan need this item? (from plan_type + rules)
|
||||
input_state: did the client supply data? (SUPPLIED | ABSENT)
|
||||
|
||||
This kills the "false green": a line is "Clear to Ship" only when EVERY required
|
||||
item is satisfied by positive evidence. A required item with no data is
|
||||
NOT_EVALUATED and blocks green — green is never granted by absence of a
|
||||
contradiction. plan_type is NEVER guessed; unknown -> "Plan Type Needed".
|
||||
|
||||
Spec: docs/readiness-model-brief-2026-06-23.md section 4.
|
||||
|
||||
Required-ness is read from payer_rules.json (single source of truth), NOT from
|
||||
duplicated Python sets, removing the latent set-vs-JSON divergence the brief
|
||||
flagged.
|
||||
|
||||
PROVISIONAL (flagged for Pi/Kisa): the mapping of gaps to the four UI labels
|
||||
(At Risk / Action Needed / On Track) below is a reasonable default. The precise
|
||||
label semantics + frontend nesting + "plan type needed" UX are a Pi design task.
|
||||
|
||||
PHI CONTRACT: operates only on plan_type and the five doc-status inputs. No
|
||||
names, SSNs, DOBs, or contact fields.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import logging
|
||||
from dataclasses import dataclass
|
||||
from datetime import date
|
||||
from enum import Enum
|
||||
from functools import lru_cache
|
||||
from pathlib import Path
|
||||
from typing import Optional
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
PAYER_RULES_PATH = Path(__file__).parent.parent / "config" / "payer_rules.json"
|
||||
|
||||
# Plan types Signal can grade. Anything else (including None or "unknown") is
|
||||
# treated as "plan type needed" — never defaulted, never guessed.
|
||||
KNOWN_PLAN_TYPES = frozenset(
|
||||
{"medicare", "medicare_advantage", "medicaid", "commercial"}
|
||||
)
|
||||
|
||||
# Universal claim/coverage requirements per LCD L33822, independent of plan type.
|
||||
UNIVERSAL_ITEMS = ("swo", "visit", "diagnosis")
|
||||
DOC_TYPES = ("swo", "visit", "pecos", "pa", "diagnosis")
|
||||
|
||||
|
||||
class RequiredState(str, Enum):
|
||||
REQUIRED = "REQUIRED"
|
||||
NOT_REQUIRED = "NOT_REQUIRED"
|
||||
NOT_EVALUATED = "NOT_EVALUATED" # cannot decide (plan type unknown)
|
||||
|
||||
|
||||
class InputState(str, Enum):
|
||||
SUPPLIED = "SUPPLIED"
|
||||
ABSENT = "ABSENT"
|
||||
|
||||
|
||||
class Quality(str, Enum):
|
||||
GOOD = "GOOD" # positive evidence the item is satisfied
|
||||
PENDING = "PENDING" # supplied but in progress (not yet good, not bad)
|
||||
BAD = "BAD" # supplied and contradicting (denied / expired / no)
|
||||
NONE = "NONE" # nothing supplied
|
||||
|
||||
|
||||
class LineStatus(str, Enum):
|
||||
CLEAR_TO_SHIP = "Clear to Ship"
|
||||
ON_TRACK = "On Track"
|
||||
ACTION_NEEDED = "Action Needed"
|
||||
AT_RISK = "At Risk"
|
||||
PLAN_TYPE_NEEDED = "Plan Type Needed"
|
||||
|
||||
|
||||
@dataclass
|
||||
class DocItem:
|
||||
doc_type: str
|
||||
required_state: RequiredState
|
||||
input_state: InputState
|
||||
quality: Quality
|
||||
value: Optional[str]
|
||||
satisfied: bool
|
||||
|
||||
|
||||
@dataclass
|
||||
class ReadinessVerdict:
|
||||
plan_type: Optional[str]
|
||||
doc_items: list[DocItem]
|
||||
line_status: LineStatus
|
||||
|
||||
|
||||
@lru_cache(maxsize=1)
|
||||
def _load_rules() -> dict:
|
||||
with open(PAYER_RULES_PATH, "r") as f:
|
||||
return json.load(f)
|
||||
|
||||
|
||||
def _norm(value: Optional[str]) -> str:
|
||||
return (value or "").strip().lower()
|
||||
|
||||
|
||||
# --- per-item quality vocabularies -----------------------------------------
|
||||
# Provisional synonym sets, fail-safe by design: an unrecognized value resolves
|
||||
# to PENDING/BAD (over-flag), never GOOD. These belong next to payer_rules.json
|
||||
# once the rules become addressable config; expand from the 25-CSV corpus.
|
||||
_SWO_GOOD = {"on file", "on-file", "active", "valid", "received", "signed", "complete"}
|
||||
_SWO_BAD = {"expired", "revoked", "cancelled", "canceled", "voided"}
|
||||
_PECOS_GOOD = {"yes", "verified", "true", "1", "enrolled", "active", "approved"}
|
||||
_PECOS_BAD = {"no", "not verified", "false", "0", "inactive", "denied", "revoked"}
|
||||
_PA_GOOD = {"approved", "not required", "authorized", "auth on file", "on file"}
|
||||
_PA_BAD = {"denied", "expired", "cancelled", "canceled", "revoked"}
|
||||
# Diagnosis carries a real ICD-10 code OR a yes/no flag, so presence = positive
|
||||
# evidence and only an EXPLICIT negative is a gap (inverted vs the others).
|
||||
_DIAGNOSIS_BAD = {"no", "missing", "none", "false", "0", "not on file", "absent", "n/a"}
|
||||
|
||||
|
||||
def _swo_quality(value: Optional[str]) -> Quality:
|
||||
s = _norm(value)
|
||||
if not s:
|
||||
return Quality.NONE
|
||||
if s in _SWO_GOOD:
|
||||
return Quality.GOOD
|
||||
if s in _SWO_BAD:
|
||||
return Quality.BAD
|
||||
return Quality.PENDING # "pending" or any other supplied state
|
||||
|
||||
|
||||
def _visit_quality(csv_visit_date: Optional[date],
|
||||
confirmed_visit_date: Optional[date]) -> Quality:
|
||||
# A confirmed/staff visit date on file is positive evidence. The estimated
|
||||
# shipment-minus-30 proxy is NOT client input and does not count here.
|
||||
# SCOPE BOUNDARY (brief section 3): this verdict checks the visit is ON FILE.
|
||||
# Whether it is still WITHIN the 6-month window is the separate timing layer
|
||||
# (CoverageFlag / CoverageLine.timing_flag), combined at the worklist level.
|
||||
if csv_visit_date or confirmed_visit_date:
|
||||
return Quality.GOOD
|
||||
return Quality.NONE
|
||||
|
||||
|
||||
def _diagnosis_quality(value: Optional[str]) -> Quality:
|
||||
s = _norm(value)
|
||||
if not s:
|
||||
return Quality.NONE
|
||||
return Quality.BAD if s in _DIAGNOSIS_BAD else Quality.GOOD
|
||||
|
||||
|
||||
def _pecos_quality(value: Optional[str]) -> Quality:
|
||||
s = _norm(value)
|
||||
if not s:
|
||||
return Quality.NONE
|
||||
if s in _PECOS_GOOD:
|
||||
return Quality.GOOD
|
||||
if s in _PECOS_BAD:
|
||||
return Quality.BAD
|
||||
return Quality.BAD # required Medicare item: unknown fails safe
|
||||
|
||||
|
||||
def _pa_quality(value: Optional[str]) -> Quality:
|
||||
s = _norm(value)
|
||||
if not s:
|
||||
return Quality.NONE
|
||||
if s in _PA_GOOD:
|
||||
return Quality.GOOD
|
||||
if s in _PA_BAD:
|
||||
return Quality.BAD
|
||||
return Quality.PENDING # "pending", "not started", etc.
|
||||
|
||||
|
||||
def _make_item(doc_type: str, required: RequiredState,
|
||||
quality: Quality, value: Optional[str]) -> DocItem:
|
||||
input_state = InputState.ABSENT if quality == Quality.NONE else InputState.SUPPLIED
|
||||
|
||||
if required == RequiredState.NOT_REQUIRED:
|
||||
satisfied = True
|
||||
elif required == RequiredState.NOT_EVALUATED:
|
||||
satisfied = False # cannot confirm -> blocks green
|
||||
else: # REQUIRED
|
||||
satisfied = quality == Quality.GOOD
|
||||
|
||||
return DocItem(
|
||||
doc_type=doc_type,
|
||||
required_state=required,
|
||||
input_state=input_state,
|
||||
quality=quality,
|
||||
value=value,
|
||||
satisfied=satisfied,
|
||||
)
|
||||
|
||||
|
||||
def _required_from_config(plan_known: bool, plan_cfg: Optional[dict],
|
||||
flag_key: str, plan_type: Optional[str]) -> RequiredState:
|
||||
"""Required-ness for a plan-dependent item, read from config.
|
||||
|
||||
A missing config (known plan absent from payer_rules.json, or the flag key
|
||||
missing) means we CANNOT decide -> NOT_EVALUATED, which blocks green. We
|
||||
never treat a config gap as 'not required' (that was the false-green hole).
|
||||
"""
|
||||
if not plan_known:
|
||||
return RequiredState.NOT_EVALUATED
|
||||
if not plan_cfg or flag_key not in plan_cfg:
|
||||
logger.warning(
|
||||
"plan_type '%s' missing '%s' in payer_rules.json -> cannot grade %s",
|
||||
plan_type, flag_key, flag_key,
|
||||
)
|
||||
return RequiredState.NOT_EVALUATED
|
||||
return RequiredState.REQUIRED if bool(plan_cfg[flag_key]) else RequiredState.NOT_REQUIRED
|
||||
|
||||
|
||||
def evaluate_readiness(
|
||||
plan_type: Optional[str],
|
||||
*,
|
||||
csv_swo_status: Optional[str] = None,
|
||||
csv_visit_date: Optional[date] = None,
|
||||
confirmed_visit_date: Optional[date] = None,
|
||||
csv_pecos_verified: Optional[str] = None,
|
||||
csv_pa_status: Optional[str] = None,
|
||||
csv_diagnosis_on_file: Optional[str] = None,
|
||||
rules: Optional[dict] = None,
|
||||
) -> ReadinessVerdict:
|
||||
"""Grade one coverage line's five documentation items and roll up a line status."""
|
||||
rules = rules if rules is not None else _load_rules()
|
||||
|
||||
# Normalize the client-mapped plan type (casing/whitespace only — never guess).
|
||||
plan_type = _norm(plan_type) or None
|
||||
plan_known = plan_type in KNOWN_PLAN_TYPES
|
||||
|
||||
# Plan-dependent required-ness from config (single source of truth, never sets).
|
||||
plan_cfg = rules.get("payer_rules", {}).get(plan_type) if plan_known else None
|
||||
pecos_required = _required_from_config(plan_known, plan_cfg, "pecos_required", plan_type)
|
||||
pa_required = _required_from_config(plan_known, plan_cfg, "pa_required", plan_type)
|
||||
|
||||
visit_date = csv_visit_date or confirmed_visit_date
|
||||
items = [
|
||||
_make_item("swo", RequiredState.REQUIRED, _swo_quality(csv_swo_status), csv_swo_status),
|
||||
_make_item("visit", RequiredState.REQUIRED,
|
||||
_visit_quality(csv_visit_date, confirmed_visit_date),
|
||||
visit_date.isoformat() if visit_date else None),
|
||||
_make_item("diagnosis", RequiredState.REQUIRED,
|
||||
_diagnosis_quality(csv_diagnosis_on_file), csv_diagnosis_on_file),
|
||||
_make_item("pecos", pecos_required, _pecos_quality(csv_pecos_verified), csv_pecos_verified),
|
||||
_make_item("pa", pa_required, _pa_quality(csv_pa_status), csv_pa_status),
|
||||
]
|
||||
|
||||
return ReadinessVerdict(
|
||||
plan_type=plan_type,
|
||||
doc_items=items,
|
||||
line_status=_line_status(items, plan_known),
|
||||
)
|
||||
|
||||
|
||||
def _line_status(items: list[DocItem], plan_known: bool) -> LineStatus:
|
||||
"""Roll the five items into one line status.
|
||||
|
||||
Clear to Ship is reachable ONLY when every required item is satisfied.
|
||||
Severity (worst first): Plan Type Needed / At Risk > Action Needed > On Track.
|
||||
|
||||
PROVISIONAL label mapping (flagged for Pi): At Risk = a supplied-and-bad
|
||||
required item (active contradiction); Action Needed = a required item with
|
||||
no data to chase; On Track = required items only in progress (pending).
|
||||
"""
|
||||
if not plan_known:
|
||||
return LineStatus.PLAN_TYPE_NEEDED
|
||||
|
||||
# A known plan whose rules are missing from config cannot be fully graded
|
||||
# (PECOS/PA NOT_EVALUATED). Fail safe: surface it, never grade it green.
|
||||
if any(i.required_state == RequiredState.NOT_EVALUATED for i in items):
|
||||
return LineStatus.PLAN_TYPE_NEEDED
|
||||
|
||||
# Green is earned only when EVERY item is satisfied (NOT_REQUIRED items are
|
||||
# satisfied by definition; required items only by positive evidence).
|
||||
if all(i.satisfied for i in items):
|
||||
return LineStatus.CLEAR_TO_SHIP
|
||||
|
||||
required = [i for i in items if i.required_state == RequiredState.REQUIRED]
|
||||
if any(i.quality == Quality.BAD for i in required):
|
||||
return LineStatus.AT_RISK
|
||||
if any(i.input_state == InputState.ABSENT for i in required):
|
||||
return LineStatus.ACTION_NEEDED
|
||||
return LineStatus.ON_TRACK
|
||||
|
|
@ -1,347 +0,0 @@
|
|||
"""
|
||||
Tests for the dedup / grouping foundation of the readiness model.
|
||||
|
||||
Source design: docs/dedup-design-2026-06-24.md (Pi, locked by Kisa 2026-06-24)
|
||||
docs/readiness-model-brief-2026-06-23.md sections 3, 5, 7
|
||||
|
||||
Locked dedup key: (patient_id, device_type, date_of_service).
|
||||
order_number is a displayed attribute, NEVER part of the dedup key.
|
||||
"""
|
||||
|
||||
import sys
|
||||
from pathlib import Path
|
||||
sys.path.insert(0, str(Path(__file__).parent.parent / "python-backend"))
|
||||
|
||||
from datetime import date, timedelta
|
||||
|
||||
from core.coverage_calculator import ShipmentRecord
|
||||
from core.dedup import (
|
||||
dedup,
|
||||
assign_to_coverage_lines,
|
||||
MergedShipment,
|
||||
CoverageLine,
|
||||
is_cgm_device,
|
||||
)
|
||||
|
||||
DOS = date(2026, 6, 1)
|
||||
|
||||
|
||||
def make_row(**kwargs):
|
||||
defaults = dict(
|
||||
patient_id="P1",
|
||||
device_type="dexcom_g7",
|
||||
shipment_date=DOS,
|
||||
quantity=3,
|
||||
payer="Medicare Part B",
|
||||
component="sensor",
|
||||
plan_type="medicare",
|
||||
)
|
||||
defaults.update(kwargs)
|
||||
return ShipmentRecord(**defaults)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Dedup: grouping by (patient_id, device_type, DOS)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_dedup_component_split():
|
||||
"""Two rows, same patient+device+DOS, different components -> one MergedShipment, summed qty."""
|
||||
rows = [
|
||||
make_row(quantity=3, component="sensor", order_number="ORD-100"),
|
||||
make_row(quantity=1, component="transmitter", order_number="ORD-101"),
|
||||
]
|
||||
result = dedup(rows)
|
||||
assert len(result) == 1
|
||||
assert result[0].total_quantity == 4
|
||||
assert set(result[0].components) == {"sensor", "transmitter"}
|
||||
assert len(result[0].order_numbers) == 2
|
||||
|
||||
|
||||
def test_dedup_identical_duplicate():
|
||||
"""Two identical rows (same component) -> max quantity, not sum (duplicate guard)."""
|
||||
rows = [
|
||||
make_row(quantity=3, component="sensor"),
|
||||
make_row(quantity=3, component="sensor"),
|
||||
]
|
||||
result = dedup(rows)
|
||||
assert len(result) == 1
|
||||
assert result[0].total_quantity == 3 # max within config, not 6
|
||||
|
||||
|
||||
def test_dedup_mixed_duplicate_and_split():
|
||||
"""Two duplicate sensor rows + one transmitter row -> max(sensor)+max(transmitter) = 4."""
|
||||
rows = [
|
||||
make_row(quantity=3, component="sensor"),
|
||||
make_row(quantity=3, component="sensor"),
|
||||
make_row(quantity=1, component="transmitter"),
|
||||
]
|
||||
result = dedup(rows)
|
||||
assert result[0].total_quantity == 4
|
||||
|
||||
|
||||
def test_dedup_zero_quantity_row_does_not_add():
|
||||
"""qty=3 sensor + qty=0 sensor -> 3 (zero row is informational)."""
|
||||
rows = [
|
||||
make_row(quantity=3, component="sensor"),
|
||||
make_row(quantity=0, component="sensor"),
|
||||
]
|
||||
result = dedup(rows)
|
||||
assert result[0].total_quantity == 3
|
||||
|
||||
|
||||
def test_dedup_wildly_different_same_component_takes_max():
|
||||
"""qty=3 + qty=100 same component -> max 100 (likely error, not a real split)."""
|
||||
rows = [
|
||||
make_row(quantity=3, component="sensor"),
|
||||
make_row(quantity=100, component="sensor"),
|
||||
]
|
||||
result = dedup(rows)
|
||||
assert result[0].total_quantity == 100
|
||||
|
||||
|
||||
def test_dedup_different_dos():
|
||||
"""Same patient+device, different DOS -> two separate shipments."""
|
||||
rows = [
|
||||
make_row(shipment_date=DOS, quantity=3),
|
||||
make_row(shipment_date=DOS + timedelta(days=30), quantity=3),
|
||||
]
|
||||
result = dedup(rows)
|
||||
assert len(result) == 2
|
||||
|
||||
|
||||
def test_dedup_different_device():
|
||||
"""Same patient+DOS, different device -> two separate shipments."""
|
||||
rows = [
|
||||
make_row(device_type="dexcom_g7", component="sensor"),
|
||||
make_row(device_type="omnipod_5", component="pod"),
|
||||
]
|
||||
result = dedup(rows)
|
||||
assert len(result) == 2
|
||||
|
||||
|
||||
def test_dedup_no_quantity_defaults_to_one():
|
||||
"""Missing quantity -> default to 1 (a row without qty is still a shipment)."""
|
||||
rows = [make_row(quantity=None, component="sensor")]
|
||||
result = dedup(rows)
|
||||
assert result[0].total_quantity == 1
|
||||
|
||||
|
||||
def test_dedup_missing_order_number():
|
||||
"""No order_number -> order_numbers list is empty, no crash."""
|
||||
rows = [make_row(order_number=None)]
|
||||
result = dedup(rows)
|
||||
assert result[0].order_numbers == []
|
||||
|
||||
|
||||
def test_dedup_order_numbers_collected_and_deduped():
|
||||
"""Repeated order numbers across rows are collected once, in first-seen order."""
|
||||
rows = [
|
||||
make_row(component="sensor", order_number="ORD-100"),
|
||||
make_row(component="transmitter", order_number="ORD-100"),
|
||||
]
|
||||
result = dedup(rows)
|
||||
assert result[0].order_numbers == ["ORD-100"]
|
||||
|
||||
|
||||
def test_dedup_hcpcs_collected():
|
||||
"""HCPCS codes collected across rows in the group."""
|
||||
rows = [
|
||||
make_row(component="sensor", hcpcs="A4239"),
|
||||
make_row(component="transmitter", hcpcs="E2103"),
|
||||
]
|
||||
result = dedup(rows)
|
||||
assert set(result[0].hcpcs_codes) == {"A4239", "E2103"}
|
||||
|
||||
|
||||
def test_dedup_payer_latest_non_null_wins():
|
||||
"""payer is display-only; latest non-null row wins."""
|
||||
rows = [
|
||||
make_row(component="sensor", payer="Old Payer"),
|
||||
make_row(component="transmitter", payer="Medicare Part B"),
|
||||
]
|
||||
result = dedup(rows)
|
||||
assert result[0].payer == "Medicare Part B"
|
||||
|
||||
|
||||
def test_dedup_csv_field_latest_non_null_wins():
|
||||
"""csv_swo_status: a later non-null value wins over an earlier null."""
|
||||
rows = [
|
||||
make_row(component="sensor", csv_swo_status=None),
|
||||
make_row(component="transmitter", csv_swo_status="On File"),
|
||||
]
|
||||
result = dedup(rows)
|
||||
assert result[0].csv_swo_status == "On File"
|
||||
|
||||
|
||||
def test_dedup_csv_field_earlier_non_null_kept_when_later_is_null():
|
||||
"""csv_swo_status: an earlier non-null is kept when the later row is null."""
|
||||
rows = [
|
||||
make_row(component="sensor", csv_swo_status="On File"),
|
||||
make_row(component="transmitter", csv_swo_status=None),
|
||||
]
|
||||
result = dedup(rows)
|
||||
assert result[0].csv_swo_status == "On File"
|
||||
|
||||
|
||||
def test_dedup_single_row_passthrough():
|
||||
"""A single row produces a MergedShipment carrying its fields."""
|
||||
rows = [make_row(quantity=2, component="sensor", order_number="ORD-1")]
|
||||
result = dedup(rows)
|
||||
assert len(result) == 1
|
||||
assert isinstance(result[0], MergedShipment)
|
||||
assert result[0].patient_id == "P1"
|
||||
assert result[0].device_type == "dexcom_g7"
|
||||
assert result[0].shipment_date == DOS
|
||||
assert result[0].total_quantity == 2
|
||||
assert result[0].plan_type == "medicare"
|
||||
|
||||
|
||||
def test_dedup_plan_type_never_derived_from_payer():
|
||||
"""plan_type carries the client-mapped value; None stays None even with a known payer name."""
|
||||
rows = [make_row(payer="Aetna PPO", plan_type=None)]
|
||||
result = dedup(rows)
|
||||
assert result[0].plan_type is None # never guessed from "Aetna"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# CoverageLine assignment: group MergedShipments into gradeable lines
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_is_cgm_device():
|
||||
assert is_cgm_device("dexcom_g7") is True
|
||||
assert is_cgm_device("freestyle_libre_3") is True
|
||||
assert is_cgm_device("omnipod_5") is False
|
||||
|
||||
|
||||
def test_assign_groups_by_patient_device_plan():
|
||||
"""One patient, one device, one plan -> one CoverageLine."""
|
||||
merged = dedup([
|
||||
make_row(shipment_date=DOS, quantity=3),
|
||||
make_row(shipment_date=DOS + timedelta(days=30), quantity=3),
|
||||
])
|
||||
lines = assign_to_coverage_lines(merged)
|
||||
assert len(lines) == 1
|
||||
|
||||
|
||||
def test_assign_latest_dos_is_current():
|
||||
"""The latest DOS shipment is the current one; older ones are prior."""
|
||||
merged = dedup([
|
||||
make_row(shipment_date=DOS, quantity=3),
|
||||
make_row(shipment_date=DOS + timedelta(days=30), quantity=3),
|
||||
])
|
||||
line = list(assign_to_coverage_lines(merged).values())[0]
|
||||
assert line.current_shipment.shipment_date == DOS + timedelta(days=30)
|
||||
assert len(line.prior_shipments) == 1
|
||||
assert line.prior_shipments[0].shipment_date == DOS
|
||||
|
||||
|
||||
def test_assign_cgm_line_is_gradeable():
|
||||
merged = dedup([make_row(device_type="dexcom_g7", component="sensor")])
|
||||
line = list(assign_to_coverage_lines(merged).values())[0]
|
||||
assert line.gradeable is True
|
||||
|
||||
|
||||
def test_assign_pump_line_not_gradeable():
|
||||
merged = dedup([make_row(device_type="omnipod_5", component="pod")])
|
||||
line = list(assign_to_coverage_lines(merged).values())[0]
|
||||
assert line.gradeable is False
|
||||
|
||||
|
||||
def test_assign_different_plan_type_separate_lines():
|
||||
"""Same patient+device, different mapped plan_type -> two CoverageLines (dual-eligible)."""
|
||||
merged = dedup([
|
||||
make_row(shipment_date=DOS, plan_type="medicare"),
|
||||
make_row(shipment_date=DOS + timedelta(days=30), plan_type="medicaid"),
|
||||
])
|
||||
lines = assign_to_coverage_lines(merged)
|
||||
assert len(lines) == 2
|
||||
|
||||
|
||||
def test_assign_none_plan_type_groups_as_unknown():
|
||||
"""plan_type None is keyed as 'unknown', not guessed, and still forms a line."""
|
||||
merged = dedup([make_row(plan_type=None)])
|
||||
lines = assign_to_coverage_lines(merged)
|
||||
assert len(lines) == 1
|
||||
line = list(lines.values())[0]
|
||||
assert line.plan_type == "unknown"
|
||||
|
||||
|
||||
def test_assign_starts_ungraded():
|
||||
"""A freshly assigned line has no status yet (gradeLine runs later)."""
|
||||
merged = dedup([make_row()])
|
||||
line = list(assign_to_coverage_lines(merged).values())[0]
|
||||
assert line.line_status is None
|
||||
assert line.timing_flag is None
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Hardening (code-auditor findings M1, L1, L3 + edge cases)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_dedup_conflicting_plan_type_routes_to_unknown():
|
||||
"""Two rows in one group with DIFFERENT non-null plan types -> never guess.
|
||||
|
||||
The plan type is grading-critical and must never be silently picked. When a
|
||||
dedup group disagrees, route to None so the line keys as 'plan type needed'.
|
||||
"""
|
||||
rows = [
|
||||
make_row(component="sensor", plan_type="medicare"),
|
||||
make_row(component="transmitter", plan_type="medicaid"),
|
||||
]
|
||||
result = dedup(rows)
|
||||
assert len(result) == 1
|
||||
assert result[0].plan_type is None
|
||||
line = list(assign_to_coverage_lines(result).values())[0]
|
||||
assert line.plan_type == "unknown"
|
||||
|
||||
|
||||
def test_dedup_agreeing_plan_type_preserved():
|
||||
"""Same non-null plan type across rows is kept (no false conflict)."""
|
||||
rows = [
|
||||
make_row(component="sensor", plan_type="medicare"),
|
||||
make_row(component="transmitter", plan_type="medicare"),
|
||||
]
|
||||
result = dedup(rows)
|
||||
assert result[0].plan_type == "medicare"
|
||||
|
||||
|
||||
def test_dedup_empty_string_csv_field_does_not_clobber():
|
||||
"""An empty-string csv value must not overwrite an earlier real value."""
|
||||
rows = [
|
||||
make_row(component="sensor", csv_swo_status="On File"),
|
||||
make_row(component="transmitter", csv_swo_status=""),
|
||||
]
|
||||
result = dedup(rows)
|
||||
assert result[0].csv_swo_status == "On File"
|
||||
|
||||
|
||||
def test_dedup_datetime_dos_coerced_to_date():
|
||||
"""Same calendar day with different timestamps -> one group (timezone edge)."""
|
||||
from datetime import datetime
|
||||
rows = [
|
||||
make_row(shipment_date=datetime(2026, 6, 1, 9, 0), component="sensor"),
|
||||
make_row(shipment_date=datetime(2026, 6, 1, 17, 0), component="transmitter"),
|
||||
]
|
||||
result = dedup(rows)
|
||||
assert len(result) == 1
|
||||
assert result[0].shipment_date == date(2026, 6, 1)
|
||||
|
||||
|
||||
def test_dedup_empty_input():
|
||||
assert dedup([]) == []
|
||||
assert assign_to_coverage_lines([]) == {}
|
||||
|
||||
|
||||
def test_cgm_device_types_exist_in_payer_rules():
|
||||
"""Drift guard: every gradeable CGM device must exist in payer_rules.json,
|
||||
and the known pump must not be in the gradeable set."""
|
||||
import json
|
||||
rules_path = (
|
||||
Path(__file__).parent.parent
|
||||
/ "python-backend" / "config" / "payer_rules.json"
|
||||
)
|
||||
devices = set(json.loads(rules_path.read_text())["devices"].keys())
|
||||
from core.dedup import CGM_DEVICE_TYPES
|
||||
missing = CGM_DEVICE_TYPES - devices
|
||||
assert not missing, f"CGM devices missing from payer_rules.json: {missing}"
|
||||
assert "omnipod_5" not in CGM_DEVICE_TYPES
|
||||
|
|
@ -1,257 +0,0 @@
|
|||
"""
|
||||
Tests for the readiness verdict engine — the "kill the false green" logic.
|
||||
|
||||
Source spec: docs/readiness-model-brief-2026-06-23.md section 4 ("The verdict rule").
|
||||
|
||||
Two axes, never conflated into one string:
|
||||
required_state: does this plan need this item? (from plan_type + rules)
|
||||
input_state: did the client supply data? (SUPPLIED | ABSENT)
|
||||
|
||||
Per DocItem:
|
||||
NOT_REQUIRED + any input -> satisfied (green-eligible)
|
||||
REQUIRED + SUPPLIED + good -> satisfied
|
||||
REQUIRED + SUPPLIED + bad -> gap
|
||||
REQUIRED + ABSENT -> NOT_EVALUATED, blocks green, never satisfied
|
||||
|
||||
A line is "Clear to Ship" ONLY if EVERY required item is satisfied. Green is
|
||||
earned by positive evidence, never granted by the absence of contradiction.
|
||||
plan_type unknown -> "Plan Type Needed", structurally cannot be green.
|
||||
"""
|
||||
|
||||
import sys
|
||||
from pathlib import Path
|
||||
sys.path.insert(0, str(Path(__file__).parent.parent / "python-backend"))
|
||||
|
||||
from datetime import date
|
||||
|
||||
from core.readiness import (
|
||||
evaluate_readiness,
|
||||
RequiredState,
|
||||
InputState,
|
||||
LineStatus,
|
||||
)
|
||||
|
||||
VISIT = date(2026, 5, 1)
|
||||
|
||||
|
||||
def items_by_type(verdict):
|
||||
return {i.doc_type: i for i in verdict.doc_items}
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# The four-rule verdict table (brief section 4)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_not_required_item_is_satisfied_even_when_absent():
|
||||
"""Medicare FFS does not require PA -> PA is satisfied with no input at all."""
|
||||
v = evaluate_readiness("medicare", csv_pa_status=None)
|
||||
pa = items_by_type(v)["pa"]
|
||||
assert pa.required_state == RequiredState.NOT_REQUIRED
|
||||
assert pa.satisfied is True
|
||||
|
||||
|
||||
def test_required_supplied_good_is_satisfied():
|
||||
v = evaluate_readiness("medicare", csv_swo_status="On File")
|
||||
swo = items_by_type(v)["swo"]
|
||||
assert swo.required_state == RequiredState.REQUIRED
|
||||
assert swo.input_state == InputState.SUPPLIED
|
||||
assert swo.satisfied is True
|
||||
|
||||
|
||||
def test_required_supplied_bad_is_gap():
|
||||
v = evaluate_readiness("medicare", csv_swo_status="Expired")
|
||||
swo = items_by_type(v)["swo"]
|
||||
assert swo.satisfied is False
|
||||
|
||||
|
||||
def test_required_absent_blocks_green():
|
||||
"""THE false-green killer: a required item with no data is never satisfied."""
|
||||
v = evaluate_readiness("medicare", csv_swo_status=None)
|
||||
swo = items_by_type(v)["swo"]
|
||||
assert swo.required_state == RequiredState.REQUIRED
|
||||
assert swo.input_state == InputState.ABSENT
|
||||
assert swo.satisfied is False
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Line status — clear to ship is earned, not defaulted
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def _all_docs_good(**overrides):
|
||||
base = dict(
|
||||
csv_swo_status="On File",
|
||||
csv_visit_date=VISIT,
|
||||
csv_pecos_verified="Yes",
|
||||
csv_pa_status="Not Required",
|
||||
csv_diagnosis_on_file="Yes",
|
||||
)
|
||||
base.update(overrides)
|
||||
return base
|
||||
|
||||
|
||||
def test_clear_to_ship_when_every_required_item_satisfied():
|
||||
v = evaluate_readiness("medicare", **_all_docs_good())
|
||||
assert v.line_status == LineStatus.CLEAR_TO_SHIP
|
||||
|
||||
|
||||
def test_missing_required_item_is_not_clear_to_ship():
|
||||
"""Old code greened this (absent column = no gap). New code must not."""
|
||||
v = evaluate_readiness("medicare", **_all_docs_good(csv_swo_status=None))
|
||||
assert v.line_status != LineStatus.CLEAR_TO_SHIP
|
||||
assert v.line_status == LineStatus.ACTION_NEEDED
|
||||
|
||||
|
||||
def test_supplied_bad_required_item_is_at_risk():
|
||||
v = evaluate_readiness("medicare_advantage", **_all_docs_good(csv_pa_status="Denied"))
|
||||
assert v.line_status == LineStatus.AT_RISK
|
||||
|
||||
|
||||
def test_pending_only_is_on_track():
|
||||
"""No hard gap, no missing data, but an item still in progress -> On Track."""
|
||||
v = evaluate_readiness("medicare", **_all_docs_good(csv_swo_status="Pending"))
|
||||
assert v.line_status == LineStatus.ON_TRACK
|
||||
|
||||
|
||||
def test_bad_outranks_absent_in_severity():
|
||||
"""A denied item (At Risk) outranks a missing item (Action Needed)."""
|
||||
v = evaluate_readiness(
|
||||
"medicare_advantage",
|
||||
**_all_docs_good(csv_pa_status="Denied", csv_diagnosis_on_file=None),
|
||||
)
|
||||
assert v.line_status == LineStatus.AT_RISK
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Plan type unknown — never guess, cannot be green
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_plan_type_none_is_plan_type_needed():
|
||||
v = evaluate_readiness(None, **_all_docs_good())
|
||||
assert v.line_status == LineStatus.PLAN_TYPE_NEEDED
|
||||
|
||||
|
||||
def test_plan_type_unknown_string_is_plan_type_needed():
|
||||
"""An unmapped plan value is treated as unknown, never defaulted/guessed."""
|
||||
v = evaluate_readiness("unknown", **_all_docs_good())
|
||||
assert v.line_status == LineStatus.PLAN_TYPE_NEEDED
|
||||
|
||||
|
||||
def test_plan_unknown_leaves_pecos_and_pa_not_evaluated():
|
||||
v = evaluate_readiness(None, **_all_docs_good())
|
||||
items = items_by_type(v)
|
||||
assert items["pecos"].required_state == RequiredState.NOT_EVALUATED
|
||||
assert items["pa"].required_state == RequiredState.NOT_EVALUATED
|
||||
|
||||
|
||||
def test_plan_unknown_still_grades_universal_items():
|
||||
"""SWO, visit, diagnosis are universal (LCD L33822) -> still graded when plan unknown."""
|
||||
v = evaluate_readiness(None, csv_swo_status="On File")
|
||||
swo = items_by_type(v)["swo"]
|
||||
assert swo.required_state == RequiredState.REQUIRED
|
||||
assert swo.satisfied is True
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Required-ness comes from payer_rules.json, not duplicated Python sets
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_pecos_required_for_medicare():
|
||||
v = evaluate_readiness("medicare", csv_pecos_verified=None)
|
||||
assert items_by_type(v)["pecos"].required_state == RequiredState.REQUIRED
|
||||
|
||||
|
||||
def test_pecos_not_required_for_medicaid():
|
||||
"""Medicaid uses state enrollment, not PECOS (deferred) -> PECOS not required."""
|
||||
v = evaluate_readiness("medicaid", csv_pecos_verified=None)
|
||||
assert items_by_type(v)["pecos"].required_state == RequiredState.NOT_REQUIRED
|
||||
|
||||
|
||||
def test_pa_required_for_medicaid():
|
||||
v = evaluate_readiness("medicaid", csv_pa_status=None)
|
||||
assert items_by_type(v)["pa"].required_state == RequiredState.REQUIRED
|
||||
|
||||
|
||||
def test_pa_not_required_for_medicare_ffs():
|
||||
v = evaluate_readiness("medicare", csv_pa_status=None)
|
||||
assert items_by_type(v)["pa"].required_state == RequiredState.NOT_REQUIRED
|
||||
|
||||
|
||||
def test_all_five_items_present():
|
||||
v = evaluate_readiness("medicare", **_all_docs_good())
|
||||
assert set(items_by_type(v).keys()) == {"swo", "visit", "pecos", "pa", "diagnosis"}
|
||||
|
||||
|
||||
def test_diagnosis_supplied_negative_is_gap():
|
||||
v = evaluate_readiness("medicare", **_all_docs_good(csv_diagnosis_on_file="No"))
|
||||
dx = items_by_type(v)["diagnosis"]
|
||||
assert dx.input_state == InputState.SUPPLIED
|
||||
assert dx.satisfied is False
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Hardening (code-auditor findings H1, H2, M1, M2, M3, M4, L1)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_known_plan_missing_config_does_not_false_green():
|
||||
"""H1/H2: a known plan absent from payer_rules.json must NOT grade green.
|
||||
|
||||
PECOS/PA cannot be evaluated without their rule config -> NOT_EVALUATED,
|
||||
which must block 'Clear to Ship'. Fails safe, never open.
|
||||
"""
|
||||
v = evaluate_readiness(
|
||||
"medicaid", rules={"payer_rules": {}},
|
||||
**_all_docs_good(csv_pa_status="Denied"),
|
||||
)
|
||||
assert v.line_status != LineStatus.CLEAR_TO_SHIP
|
||||
assert items_by_type(v)["pa"].required_state == RequiredState.NOT_EVALUATED
|
||||
assert items_by_type(v)["pecos"].required_state == RequiredState.NOT_EVALUATED
|
||||
|
||||
|
||||
def test_diagnosis_icd10_code_counts_as_on_file():
|
||||
"""M2: a real ICD-10 code in the diagnosis field is positive evidence, not a gap."""
|
||||
v = evaluate_readiness("medicare", **_all_docs_good(csv_diagnosis_on_file="E11.9"))
|
||||
assert items_by_type(v)["diagnosis"].satisfied is True
|
||||
|
||||
|
||||
def test_diagnosis_explicit_negative_is_bad():
|
||||
v = evaluate_readiness("medicare", **_all_docs_good(csv_diagnosis_on_file="Missing"))
|
||||
assert items_by_type(v)["diagnosis"].satisfied is False
|
||||
|
||||
|
||||
def test_pecos_enrolled_synonym_is_good():
|
||||
"""M3: standard PECOS portal statuses should read as verified."""
|
||||
v = evaluate_readiness("medicare", **_all_docs_good(csv_pecos_verified="Enrolled"))
|
||||
assert items_by_type(v)["pecos"].satisfied is True
|
||||
|
||||
|
||||
def test_swo_active_synonym_is_good():
|
||||
v = evaluate_readiness("medicare", **_all_docs_good(csv_swo_status="Active"))
|
||||
assert items_by_type(v)["swo"].satisfied is True
|
||||
|
||||
|
||||
def test_plan_type_case_and_whitespace_normalized():
|
||||
"""M4: a correctly-mapped plan with stray case/space must still grade."""
|
||||
v = evaluate_readiness(" Medicare ", **_all_docs_good())
|
||||
assert v.line_status == LineStatus.CLEAR_TO_SHIP
|
||||
|
||||
|
||||
def test_staff_confirmed_visit_carries_value_for_citation():
|
||||
"""L1: a staff-confirmed visit (no CSV date) still records its date for the citation."""
|
||||
v = evaluate_readiness(
|
||||
"medicare", confirmed_visit_date=VISIT,
|
||||
csv_swo_status="On File", csv_pa_status="Not Required",
|
||||
csv_diagnosis_on_file="Yes", csv_pecos_verified="Yes",
|
||||
)
|
||||
assert items_by_type(v)["visit"].value == VISIT.isoformat()
|
||||
|
||||
|
||||
def test_known_plan_types_match_payer_rules_config():
|
||||
"""M1 drift guard: the gradeable allow-list must equal the configured plans."""
|
||||
import json
|
||||
rules_path = (
|
||||
Path(__file__).parent.parent
|
||||
/ "python-backend" / "config" / "payer_rules.json"
|
||||
)
|
||||
configured = set(json.loads(rules_path.read_text())["payer_rules"].keys()) - {"default"}
|
||||
from core.readiness import KNOWN_PLAN_TYPES
|
||||
assert set(KNOWN_PLAN_TYPES) == configured
|
||||
Loading…
Reference in a new issue