feat: citation rule IDs on every documentation requirement (plan 02, P2)
core/rules.py format-tolerant readers + stable registry (R001-R004 plan-scoped, R100-R102 universal per LCD L33822). payer_rules.json wrapped in the same commit as both engine readers (coupling hazard closed). Every DocItem now cites the rule that decided it; NOT_EVALUATED cites nothing. Independent audit: SHIP (registry-drift test added per its L2 finding). 138 tests green. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Kisa
parent
242e54e01a
commit
d24340c47a
8 changed files with 216 additions and 37 deletions
|
|
@ -180,6 +180,9 @@ class ReadinessItemOut(BaseModel):
|
|||
quality: str # GOOD | PENDING | BAD | NONE
|
||||
value: Optional[str] = None
|
||||
satisfied: bool
|
||||
# Citation: the payer_rules.json rule that decided required_state.
|
||||
# None when NOT_EVALUATED (no rule fired, nothing to cite).
|
||||
rule_id: Optional[str] = None
|
||||
|
||||
|
||||
class RecordOut(BaseModel):
|
||||
|
|
@ -427,6 +430,7 @@ def _to_record_out(
|
|||
quality=i.quality.value,
|
||||
value=i.value,
|
||||
satisfied=i.satisfied,
|
||||
rule_id=i.rule_id,
|
||||
)
|
||||
for i in readiness.doc_items
|
||||
]
|
||||
|
|
|
|||
|
|
@ -1,5 +1,14 @@
|
|||
{
|
||||
"_comment": "Wear-day rules by device type and payer. Updated 2026-06-07. Run quarterly review per compliance checklist.",
|
||||
"_comment": "Wear-day rules by device type and payer. Updated 2026-07-07 (P2: plan-scoped keys wrapped with citation rule_ids). Run quarterly review per compliance checklist.",
|
||||
"_rule_registry": {
|
||||
"R001": "visit_renewal_days — visit renewal window in days (standard: 180)",
|
||||
"R002": "refill_window_days — resupply window in days before coverage end (standard: 30; legacy code key name retained)",
|
||||
"R003": "pa_required — prior authorization required for this plan type",
|
||||
"R004": "pecos_required — PECOS enrollment verification required for this plan type",
|
||||
"R100": "swo_required — SWO is a universal claim requirement per LCD L33822",
|
||||
"R101": "visit_required — qualifying visit is a universal requirement per LCD L33822",
|
||||
"R102": "diagnosis_required — diagnosis (ICD-10) is a universal requirement per LCD L33822"
|
||||
},
|
||||
"devices": {
|
||||
"dexcom_g6": {
|
||||
"display_name": "Dexcom G6",
|
||||
|
|
@ -31,42 +40,42 @@
|
|||
},
|
||||
"payer_rules": {
|
||||
"medicare": {
|
||||
"visit_renewal_days": 180,
|
||||
"refill_window_days": 30,
|
||||
"pa_required": false,
|
||||
"pecos_required": true,
|
||||
"visit_renewal_days": { "value": 180, "rule_id": "R001" },
|
||||
"refill_window_days": { "value": 30, "rule_id": "R002" },
|
||||
"pa_required": { "value": false, "rule_id": "R003" },
|
||||
"pecos_required": { "value": true, "rule_id": "R004" },
|
||||
"_note": "Medicare FFS. PA not required (CGM coverage expansion 2023). Face-to-face visit required every 6 months. CMS-1828-F: suppliers at >=90% affirmation rate may qualify for PA exemption from June 1, 2026.",
|
||||
"covered_devices": ["dexcom_g6", "dexcom_g7", "freestyle_libre_2", "freestyle_libre_3"]
|
||||
},
|
||||
"medicare_advantage": {
|
||||
"visit_renewal_days": 180,
|
||||
"refill_window_days": 30,
|
||||
"pa_required": true,
|
||||
"pecos_required": true,
|
||||
"visit_renewal_days": { "value": 180, "rule_id": "R001" },
|
||||
"refill_window_days": { "value": 30, "rule_id": "R002" },
|
||||
"pa_required": { "value": true, "rule_id": "R003" },
|
||||
"pecos_required": { "value": true, "rule_id": "R004" },
|
||||
"_note": "Medicare Advantage — PA required, varies by plan. Verify plan-specific rules.",
|
||||
"covered_devices": ["dexcom_g6", "dexcom_g7", "freestyle_libre_2", "freestyle_libre_3"]
|
||||
},
|
||||
"medicaid": {
|
||||
"visit_renewal_days": 180,
|
||||
"refill_window_days": 30,
|
||||
"pa_required": true,
|
||||
"pecos_required": false,
|
||||
"visit_renewal_days": { "value": 180, "rule_id": "R001" },
|
||||
"refill_window_days": { "value": 30, "rule_id": "R002" },
|
||||
"pa_required": { "value": true, "rule_id": "R003" },
|
||||
"pecos_required": { "value": false, "rule_id": "R004" },
|
||||
"_note": "PA required — Keystone First, UPMC Health Plan, Highmark BCBS require PA. NJ FamilyCare, Horizon BCBS NJ require PA.",
|
||||
"covered_devices": []
|
||||
},
|
||||
"commercial": {
|
||||
"visit_renewal_days": 180,
|
||||
"refill_window_days": 30,
|
||||
"pa_required": true,
|
||||
"pecos_required": false,
|
||||
"visit_renewal_days": { "value": 180, "rule_id": "R001" },
|
||||
"refill_window_days": { "value": 30, "rule_id": "R002" },
|
||||
"pa_required": { "value": true, "rule_id": "R003" },
|
||||
"pecos_required": { "value": false, "rule_id": "R004" },
|
||||
"_note": "Commercial payer rules vary by plan. PA required by default — verify per plan.",
|
||||
"covered_devices": []
|
||||
},
|
||||
"default": {
|
||||
"visit_renewal_days": 180,
|
||||
"refill_window_days": 30,
|
||||
"pa_required": true,
|
||||
"pecos_required": false,
|
||||
"visit_renewal_days": { "value": 180, "rule_id": "R001" },
|
||||
"refill_window_days": { "value": 30, "rule_id": "R002" },
|
||||
"pa_required": { "value": true, "rule_id": "R003" },
|
||||
"pecos_required": { "value": false, "rule_id": "R004" },
|
||||
"covered_devices": []
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -328,8 +328,10 @@ def calculate_coverage(
|
|||
rule_version=RULE_VERSION,
|
||||
)
|
||||
|
||||
visit_renewal_days = payer_config.get("visit_renewal_days")
|
||||
refill_window_days = payer_config.get("refill_window_days", 30)
|
||||
from core.rules import rule_value
|
||||
|
||||
visit_renewal_days = rule_value(payer_config, "visit_renewal_days")
|
||||
refill_window_days = rule_value(payer_config, "refill_window_days", 30)
|
||||
|
||||
# Resolve visit date using priority chain
|
||||
visit_date, confidence = _resolve_visit_date(record, confirmed_visit_date)
|
||||
|
|
|
|||
|
|
@ -38,6 +38,8 @@ from functools import lru_cache
|
|||
from pathlib import Path
|
||||
from typing import Optional
|
||||
|
||||
from core.rules import UNIVERSAL_RULE_IDS, rule_id_for, rule_value
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
PAYER_RULES_PATH = Path(__file__).parent.parent / "config" / "payer_rules.json"
|
||||
|
|
@ -87,6 +89,9 @@ class DocItem:
|
|||
quality: Quality
|
||||
value: Optional[str]
|
||||
satisfied: bool
|
||||
# Names the payer_rules.json rule that decided required_state. None when
|
||||
# required_state is NOT_EVALUATED: no rule fired, so nothing is cited.
|
||||
rule_id: Optional[str] = None
|
||||
|
||||
|
||||
@dataclass
|
||||
|
|
@ -174,7 +179,8 @@ def _pa_quality(value: Optional[str]) -> Quality:
|
|||
|
||||
|
||||
def _make_item(doc_type: str, required: RequiredState,
|
||||
quality: Quality, value: Optional[str]) -> DocItem:
|
||||
quality: Quality, value: Optional[str],
|
||||
rule_id: Optional[str] = None) -> DocItem:
|
||||
input_state = InputState.ABSENT if quality == Quality.NONE else InputState.SUPPLIED
|
||||
|
||||
if required == RequiredState.NOT_REQUIRED:
|
||||
|
|
@ -191,26 +197,36 @@ def _make_item(doc_type: str, required: RequiredState,
|
|||
quality=quality,
|
||||
value=value,
|
||||
satisfied=satisfied,
|
||||
rule_id=rule_id,
|
||||
)
|
||||
|
||||
|
||||
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.
|
||||
def _required_from_config(
|
||||
plan_known: bool, plan_cfg: Optional[dict],
|
||||
flag_key: str, plan_type: Optional[str],
|
||||
) -> tuple[RequiredState, Optional[str]]:
|
||||
"""(Required-ness, citing rule_id) for a plan-dependent item, 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).
|
||||
When NOT_EVALUATED, rule_id is None: no rule fired, nothing is cited.
|
||||
Accepts both flat and wrapped ({"value":..., "rule_id":...}) config values.
|
||||
"""
|
||||
if not plan_known:
|
||||
return RequiredState.NOT_EVALUATED
|
||||
return RequiredState.NOT_EVALUATED, None
|
||||
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
|
||||
return RequiredState.NOT_EVALUATED, None
|
||||
required = (
|
||||
RequiredState.REQUIRED
|
||||
if bool(rule_value(plan_cfg, flag_key))
|
||||
else RequiredState.NOT_REQUIRED
|
||||
)
|
||||
return required, rule_id_for(plan_cfg, flag_key)
|
||||
|
||||
|
||||
def evaluate_readiness(
|
||||
|
|
@ -233,19 +249,24 @@ def evaluate_readiness(
|
|||
|
||||
# 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)
|
||||
pecos_required, pecos_rule = _required_from_config(plan_known, plan_cfg, "pecos_required", plan_type)
|
||||
pa_required, pa_rule = _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("swo", RequiredState.REQUIRED, _swo_quality(csv_swo_status),
|
||||
csv_swo_status, rule_id=UNIVERSAL_RULE_IDS["swo"]),
|
||||
_make_item("visit", RequiredState.REQUIRED,
|
||||
_visit_quality(csv_visit_date, confirmed_visit_date),
|
||||
visit_date.isoformat() if visit_date else None),
|
||||
visit_date.isoformat() if visit_date else None,
|
||||
rule_id=UNIVERSAL_RULE_IDS["visit"]),
|
||||
_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),
|
||||
_diagnosis_quality(csv_diagnosis_on_file), csv_diagnosis_on_file,
|
||||
rule_id=UNIVERSAL_RULE_IDS["diagnosis"]),
|
||||
_make_item("pecos", pecos_required, _pecos_quality(csv_pecos_verified),
|
||||
csv_pecos_verified, rule_id=pecos_rule),
|
||||
_make_item("pa", pa_required, _pa_quality(csv_pa_status),
|
||||
csv_pa_status, rule_id=pa_rule),
|
||||
]
|
||||
|
||||
return ReadinessVerdict(
|
||||
|
|
|
|||
52
python-backend/core/rules.py
Normal file
52
python-backend/core/rules.py
Normal file
|
|
@ -0,0 +1,52 @@
|
|||
"""
|
||||
rules.py
|
||||
Signal — STTIL Solutions
|
||||
|
||||
Format-tolerant readers for payer_rules.json plus the stable rule-ID registry
|
||||
(build plan 02, item P2; design: readiness-model-design-phase2 section 3.1).
|
||||
|
||||
A rule value may be FLAT (legacy: `"visit_renewal_days": 180`) or WRAPPED with
|
||||
its citation (`"visit_renewal_days": {"value": 180, "rule_id": "R001"}`).
|
||||
Both engines read through these helpers so the JSON format can carry citation
|
||||
IDs without breaking either reader.
|
||||
|
||||
RULE IDS ARE STABLE — never change an ID after shipping. The ID names the rule
|
||||
KIND; the plan type plus the value together are the citation.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Optional
|
||||
|
||||
# Universal claim/coverage requirements per LCD L33822, independent of plan
|
||||
# type. They live in code (no plan-type dependency) but carry rule IDs for
|
||||
# citation consistency.
|
||||
UNIVERSAL_RULE_IDS = {"swo": "R100", "visit": "R101", "diagnosis": "R102"}
|
||||
|
||||
# Plan-scoped keys in payer_rules.json and their registry IDs.
|
||||
PLAN_KEY_RULE_IDS = {
|
||||
"visit_renewal_days": "R001",
|
||||
"refill_window_days": "R002",
|
||||
"pa_required": "R003",
|
||||
"pecos_required": "R004",
|
||||
}
|
||||
|
||||
|
||||
def rule_value(cfg: Optional[dict], key: str, default=None):
|
||||
"""Read a rule value, accepting flat (180) and wrapped
|
||||
({'value': 180, 'rule_id': 'R001'}) formats."""
|
||||
if not cfg:
|
||||
return default
|
||||
raw = cfg.get(key, default)
|
||||
if isinstance(raw, dict) and "value" in raw:
|
||||
return raw["value"]
|
||||
return raw
|
||||
|
||||
|
||||
def rule_id_for(cfg: Optional[dict], key: str) -> Optional[str]:
|
||||
"""The citation ID for a plan-scoped rule: from the wrapped config when
|
||||
present, else the registry default for that key."""
|
||||
raw = cfg.get(key) if cfg else None
|
||||
if isinstance(raw, dict) and "rule_id" in raw:
|
||||
return raw["rule_id"]
|
||||
return PLAN_KEY_RULE_IDS.get(key)
|
||||
|
|
@ -101,3 +101,12 @@ def test_priority_sort_order():
|
|||
assert results[0].flag == CoverageFlag.SUPPLY_LAPSED
|
||||
assert results[1].flag == CoverageFlag.RENEWAL_CRITICAL
|
||||
assert results[2].flag == CoverageFlag.ACTIVE
|
||||
|
||||
|
||||
def test_wrapped_config_drives_timing_rules():
|
||||
"""P2: with the wrapped citation format on disk, the timing calculator
|
||||
still resolves visit_renewal_days correctly (R001 = 180 days)."""
|
||||
visit = TODAY - timedelta(days=30)
|
||||
r = make_record(csv_visit_date=visit)
|
||||
result = calculate_coverage(r)
|
||||
assert result.next_visit_due_date == visit + timedelta(days=180)
|
||||
|
|
|
|||
|
|
@ -255,3 +255,70 @@ def test_known_plan_types_match_payer_rules_config():
|
|||
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
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# P2: citation rule IDs (build plan 02, readiness-model-design-phase2 §3.1)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_doc_items_carry_rule_ids():
|
||||
v = evaluate_readiness(
|
||||
"medicare",
|
||||
csv_swo_status="On File",
|
||||
csv_visit_date=VISIT,
|
||||
csv_pecos_verified="Yes",
|
||||
csv_diagnosis_on_file="Yes",
|
||||
)
|
||||
items = items_by_type(v)
|
||||
assert items["swo"].rule_id == "R100"
|
||||
assert items["visit"].rule_id == "R101"
|
||||
assert items["diagnosis"].rule_id == "R102"
|
||||
assert items["pecos"].rule_id == "R004"
|
||||
assert items["pa"].rule_id == "R003"
|
||||
|
||||
|
||||
def test_flat_and_wrapped_config_produce_identical_verdicts():
|
||||
flat = {"payer_rules": {"medicare": {
|
||||
"pa_required": False, "pecos_required": True,
|
||||
}}}
|
||||
wrapped = {"payer_rules": {"medicare": {
|
||||
"pa_required": {"value": False, "rule_id": "R003"},
|
||||
"pecos_required": {"value": True, "rule_id": "R004"},
|
||||
}}}
|
||||
kwargs = dict(
|
||||
csv_swo_status="On File", csv_visit_date=VISIT,
|
||||
csv_pecos_verified="Yes", csv_diagnosis_on_file="Yes",
|
||||
)
|
||||
v_flat = evaluate_readiness("medicare", rules=flat, **kwargs)
|
||||
v_wrapped = evaluate_readiness("medicare", rules=wrapped, **kwargs)
|
||||
assert v_flat.line_status == v_wrapped.line_status
|
||||
for a, b in zip(v_flat.doc_items, v_wrapped.doc_items):
|
||||
assert (a.doc_type, a.required_state, a.satisfied) == (
|
||||
b.doc_type, b.required_state, b.satisfied
|
||||
)
|
||||
|
||||
|
||||
def test_not_evaluated_items_have_no_rule_id():
|
||||
v = evaluate_readiness(None)
|
||||
items = items_by_type(v)
|
||||
assert items["pecos"].required_state == RequiredState.NOT_EVALUATED
|
||||
assert items["pecos"].rule_id is None
|
||||
assert items["pa"].rule_id is None
|
||||
|
||||
|
||||
def test_on_disk_rule_ids_match_registry():
|
||||
"""Guards citation drift: every wrapped rule_id in payer_rules.json must
|
||||
equal the registry ID for its key (auditor finding L2, 2026-07-07)."""
|
||||
import json
|
||||
from pathlib import Path as _P
|
||||
from core.rules import PLAN_KEY_RULE_IDS
|
||||
|
||||
cfg = json.loads(
|
||||
(_P(__file__).parent.parent / "python-backend" / "config" / "payer_rules.json").read_text()
|
||||
)
|
||||
for plan, entry in cfg["payer_rules"].items():
|
||||
for key, expected_id in PLAN_KEY_RULE_IDS.items():
|
||||
raw = entry.get(key)
|
||||
assert isinstance(raw, dict) and raw.get("rule_id") == expected_id, (
|
||||
f"{plan}.{key} must be wrapped with rule_id {expected_id}"
|
||||
)
|
||||
|
|
|
|||
|
|
@ -594,3 +594,18 @@ def test_confirm_visit_normalizes_plan_type(monkeypatch):
|
|||
rec = resp.json()
|
||||
assert rec["plan_type"] == "medicare"
|
||||
assert rec["readiness_status"] == "Clear to Ship"
|
||||
|
||||
|
||||
def test_upload_readiness_items_carry_rule_ids():
|
||||
"""P2 contract W2: the serialized doc_items cite their rules."""
|
||||
csv_text = (
|
||||
HEADER
|
||||
+ f"\nPT-RID-1,Dexcom G7,{SHIP},3,Medicare Part B,medicare,"
|
||||
+ f"On File,{VISIT},Yes,,Yes"
|
||||
)
|
||||
resp = _upload(csv_text)
|
||||
assert resp.status_code == 200
|
||||
items = {i["doc_type"]: i for i in resp.json()["records"][0]["readiness_items"]}
|
||||
assert items["pecos"]["rule_id"] == "R004"
|
||||
assert items["pa"]["rule_id"] == "R003"
|
||||
assert items["swo"]["rule_id"] == "R100"
|
||||
|
|
|
|||
Loading…
Reference in a new issue