From d24340c47aa161726d2a8669b9d4c5abd857fb27 Mon Sep 17 00:00:00 2001 From: Kisa Date: Tue, 7 Jul 2026 05:10:52 -0400 Subject: [PATCH] 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 --- python-backend/api/main.py | 4 ++ python-backend/config/payer_rules.json | 51 +++++++++------- python-backend/core/coverage_calculator.py | 6 +- python-backend/core/readiness.py | 49 +++++++++++----- python-backend/core/rules.py | 52 +++++++++++++++++ tests/test_coverage_flags.py | 9 +++ tests/test_readiness.py | 67 ++++++++++++++++++++++ tests/test_worklist_readiness.py | 15 +++++ 8 files changed, 216 insertions(+), 37 deletions(-) create mode 100644 python-backend/core/rules.py diff --git a/python-backend/api/main.py b/python-backend/api/main.py index 56bacc9..13a7ba8 100644 --- a/python-backend/api/main.py +++ b/python-backend/api/main.py @@ -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 ] diff --git a/python-backend/config/payer_rules.json b/python-backend/config/payer_rules.json index fb28354..8a379da 100644 --- a/python-backend/config/payer_rules.json +++ b/python-backend/config/payer_rules.json @@ -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": [] } } diff --git a/python-backend/core/coverage_calculator.py b/python-backend/core/coverage_calculator.py index 8085d0e..1abb72b 100644 --- a/python-backend/core/coverage_calculator.py +++ b/python-backend/core/coverage_calculator.py @@ -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) diff --git a/python-backend/core/readiness.py b/python-backend/core/readiness.py index ca5444a..dc53512 100644 --- a/python-backend/core/readiness.py +++ b/python-backend/core/readiness.py @@ -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( diff --git a/python-backend/core/rules.py b/python-backend/core/rules.py new file mode 100644 index 0000000..ad3bc67 --- /dev/null +++ b/python-backend/core/rules.py @@ -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) diff --git a/tests/test_coverage_flags.py b/tests/test_coverage_flags.py index 319101a..ab4b5ef 100644 --- a/tests/test_coverage_flags.py +++ b/tests/test_coverage_flags.py @@ -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) diff --git a/tests/test_readiness.py b/tests/test_readiness.py index f598939..83ee2bc 100644 --- a/tests/test_readiness.py +++ b/tests/test_readiness.py @@ -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}" + ) diff --git a/tests/test_worklist_readiness.py b/tests/test_worklist_readiness.py index 14e5f3e..7abcca7 100644 --- a/tests/test_worklist_readiness.py +++ b/tests/test_worklist_readiness.py @@ -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"