sttil/Signal
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Compare commits

base: sttil:242e54e01a38827ecb81e16d234e76cb51f5f4dc
Branches Tags
sttil:main
..
compare: sttil:77395a1112924a4fb70940a8600e87b1b2f3c896
Branches Tags
sttil:main

No commits in common. "242e54e01a38827ecb81e16d234e76cb51f5f4dc" and "77395a1112924a4fb70940a8600e87b1b2f3c896" have entirely different histories.
242e54e01a ... 77395a1112

7 changed files with 25 additions and 948 deletions
Show stats Expand all files Collapse all files

7
context/current-state.md
View file

@ -2,9 +2,9 @@ v:1
# current-state
ACTIVE: 2026-07-07 [CC]. SIGNAL BUILD UN-PAUSED — readiness-model completion in flight per docs/build-plans/. P1 (wiring) SHIPPED + LIVE-VERIFIED (commit a3af499, deployed to Railway, false-green kill observed live). Executing P2 (citation IDs) -> P3 (rollup) -> P4 (device overrides) -> P5 (plan-type enforcement) next, one item at a time with independent audit each. P6 (frontend nesting) BLOCKED on Kisa approving the rendered mockup (built, awaiting render+review).
ACTIVE: 2026-07-06 (AIOS sessions, not a Signal build). Session 1: Crew Compounding Memory Phase 1 built + independently verified PASS; UMSIS ruled docs-only. Session 2: model routing SETTLED (strongest model orchestrates in a plain full-MCP session, sonnet/haiku subagents execute; rubric captured in operating-rules.md), watcher fleet audited healthy, research watcher confirmed never built. FABLE WINDOW ENDS 2026-07-07 09:00 (com.sttil.cos-model-switch swaps zshrc to opus-4-8). Signal build itself UNCHANGED since the 2026-06-28 checkpoint below.
NEXT: (CC) P2-P5 per docs/build-plans/02-readiness-model-completion.md; (Kisa) approve/reject P6 mockup decisions A-D; (Kisa) confirm OQ1 plan_type-ingestion decision recorded 2026-07-07 (shipped per umbrella plan; two-line revert if reversed); (Kisa) attorney email bundle + Gaboro re-approach per reactivation plan 00 (unchanged). Queued behind Signal: UMSIS docs-only activation, "63% written off" Draft fix.
NEXT: Kisa ramps her other priorities first. Queued: execute the UMSIS docs-only repo activation + cto-architect build plan; do the UMSIS KG add (present conflict, get Kisa confirm); place the domain work-plan doc in the vault; fix the retired-stat "63% written off" Draft before publish; Signal readiness model still queued from the 2026-06-27 checkpoint. PARKED (agreed 2026-07-06, build later): apply the model-routing one-liner to global CLAUDE.md + statusline model display; optional TriLane research watcher (viable on the proven launchd rails; gate via integration-steward + a curation filter).
PRIOR ACTIVE: 2026-06-28. Signal paused at clean checkpoint — Insight Engine in active BUILD phase for July 4 deadline. See insight-engine/context/current-state.md and insight-engine/handoff-clock.md for full state. Key progress today: x-sight guided experience designed (Compass) and approved by Kisa, lane-lock handoff clock created, frontend finish readiness assessed (15 ranked items). Claude is building the Compass landing + B1 deploy blocker + finish items. Signal READINESS MODEL still queued at the clean checkpoint from 2026-06-27.
@ -25,7 +25,6 @@ Insight Engine (July 4 deadline = tomorrow) is the active project. Key files:
## Decisions (last
- 2026-07-07 [CC]: Readiness model WIRED LIVE (plan 01, commit a3af499, Railway deploy verified). Phase 2a seam confirmed: readiness AUGMENTS legacy output (byte-identical legacy proven by live diff). OQ1 executed per umbrella plan 00 ("ship plan_type ingestion with the wiring") — minimal client-mapped column, never guessed; flagged for Kisa's explicit confirmation, two-line revert if reversed. Adversarial review caught a false-green in the draft row-resolution; fixed via dedup-group-keyed ReadinessIndex, regression-locked. Grounded test count: 132 (the old "73 tests" figure corrected per plan 01 §12.5).
- 2026-07-06 [CC]: Model routing SETTLED (Kisa agreed) — strongest model in the orchestrator seat (plain claude, full MCP, skip-permissions), sonnet/haiku subagents for mechanical work, Fable-as-subagent only for cleanly separable hard problems. Docs-verified: delegation multiplies tokens on one shared pool; per-agent model override + mid-session /model GROUNDED. Config application PARKED. Discovered com.sttil.cos-model-switch: Fable-to-Opus swap fires 2026-07-07 09:00. Watcher audit: 9 agents healthy, clean exits; research watcher never built.
- 2026-07-06 [CC]: Crew Compounding Memory Phase 1 shipped + independently verified — retrieve-and-inject wired into the dispatch skill (validity-stamped "WHAT THE CREW ALREADY KNOWS" block, current==true filter, recall guard) + ~/.claude/crew-log.md; built by the memory-crew, verified PASS (2 agents flagged every planted contradiction, independent grader confirmed). Also ruled: UMSIS activates sttil/umsis docs-only (state in the repo, not the vault). Spec + build record: STTIL-Vault/Projects/STTIL Solutions/Crew Compounding Memory - Design Spec - 2026-07-05.md.
- 2026-06-27 [CC]: Readiness engines COMMITTED — c07d054 (dedup foundation) + 206bf71 (verdict engine); code-auditor found + fixed HIGH config-surface false-green holes (a known plan missing from payer_rules.json no longer grades green; the green gate now checks every item); 73 tests green. Pi readiness-UX brief staged. labABLE CRUSH brief SENT. PRIORITIZATION DECISION (Kisa): Signal active build pauses at this checkpoint; Claude + Kisa finish the Insight Engine to its July 4 deadline (separate lane); Pi advances the Signal UX design async. Lanes-clear: one project per terminal/session.
@ -58,4 +57,4 @@ Insight Engine (July 4 deadline = tomorrow) is the active project. Key files:
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-07-07 (CC: Signal un-paused; P1 readiness wiring shipped+live-verified a3af499; P2-P5 executing; P6 mockup awaiting Kisa; OQ1 recorded for confirmation.)
## Updated: 2026-07-06 (CC: second AIOS session — model routing settled + captured to operating-rules.md; watcher fleet audited clean; research watcher confirmed not built; Fable window ends 2026-07-07 09:00. Signal build unchanged since 2026-06-28.)

9
docs/build-plans/01-readiness-model-wiring.md
View file

@ -313,15 +313,6 @@ records ──┬─> calculate_batch ──────────────
## 7. Skeleton snippets (illustrative only, do NOT paste blindly; re-verify line numbers at execution time)
> **EXECUTION CORRECTION (2026-07-07, found in adversarial review):** the sketches
> below key row-to-verdict resolution off `rec.plan_type` (the row's OWN value).
> That is WRONG when a row's plan_type differs from its dedup group's MERGED
> plan_type (conflicting groups merge to None) — it can attach another line's
> verdict, a reachable false green. The shipped implementation resolves rows
> through the dedup group key `(patient_id, device_type, DOS-as-date)` via
> `ReadinessIndex` in `core/worklist_readiness.py`. Follow the shipped code,
> not these sketches.
### `core/worklist_readiness.py` (new)
```python

26
docs/ship-status-ledger.md
View file

@ -59,29 +59,3 @@ Pi's dedup design (item 9) and Convex-spike design (item 6) were verified line-b
- **dedup-design-2026-06-24.md**: (1) added a separate client-mapped `plan_type` field, distinct from display `payer`, so CoverageLine keys on the mapped plan type, never the payer name — the original draft reintroduced the killed plan-type-guessing bug. (2) Fixed `merge_quantities` to sum across distinct component configs; the old blanket-max dropped quantity in mixed duplicate+split groups (e.g. two [sensor] + one [transmitter] returned 3 instead of 4).
- **convex-signal-spike-design-2026-06-24.md**: `shipments.orderNumber`/`hcpcs` changed to arrays (post-dedup multi-value); §11 dedup note de-staled to LOCKED; carry-forward note to grade on the mapped `planType`.
- Noted, not a blocker: "Convex auth -> Clerk = config-only swap" is optimistic; treat as low-but-nonzero effort (Convex Auth still maturing).
## Shipped: readiness model wired live (2026-07-07, Claude, commit a3af499)
Build plan 01 executed end to end. Phase 2a coexistence: readiness verdicts ride
additively on `/api/upload` and `/api/confirm-visit`; legacy output proven
byte-identical (live diff vs pre-deploy baseline: NONE).
- New `core/worklist_readiness.py` (ReadinessIndex). Row-to-line membership
resolves through the dedup group key `(patient_id, device_type, DOS)` — an
adversarial review reproduced a false green in the first draft's row-keyed
resolution; fixed and regression-locked before ship.
- Minimal plan_type CSV mapping shipped with the wiring (per umbrella plan 00;
OQ1 recorded for Kisa's explicit confirmation). Values lowercased+trimmed
only, never guessed; unknown values grade Plan Type Needed.
- Pre-existing bug fixed: record_lookup missed for ALL order-numbered CSVs
(CoverageResult has no order_number), silently dropping doc_state. Fallback
now resolves through the dedup MergedShipment (latest-non-null fields,
collected order numbers).
- Verification (measured, live): 132 tests green; Pi E2E 33/33; adversarial
3-lens review + refutation (7 confirmed findings, all fixed); independent
code-auditor SHIP. Live: no-plan_type upload = 3/3 "Plan Type Needed",
stats reconcile; engineered false-green row reads legacy "Clear to Ship" +
readiness "Action Needed" in the same record; confirm-visit flips verdict to
Clear to Ship and normalizes "Medicare" -> "medicare".
- Frontend intentionally renders nothing new yet (P6, gated on Kisa's mockup
approval).

94
python-backend/api/main.py
View file

@ -35,13 +35,6 @@ from core.persistence import (
load_doc_statuses_for_org,
)
from api.normalizer import normalize_csv
from core.worklist_readiness import (
build_readiness_index,
merged_to_record,
readiness_for_record,
readiness_stats,
resolve_group,
)
app = FastAPI(title="Signal API", version="1.0.0", docs_url="/docs")
@ -173,15 +166,6 @@ class DocStateOut(BaseModel):
diagnosis: str
class ReadinessItemOut(BaseModel):
doc_type: str
required_state: str # REQUIRED | NOT_REQUIRED | NOT_EVALUATED
input_state: str # SUPPLIED | ABSENT
quality: str # GOOD | PENDING | BAD | NONE
value: Optional[str] = None
satisfied: bool
class RecordOut(BaseModel):
patient_id: str
device_type: str
@ -215,11 +199,6 @@ class RecordOut(BaseModel):
csv_pa_status: Optional[str] = None
csv_diagnosis_on_file: Optional[str] = None
csv_transfer_from: Optional[str] = None
# Readiness verdict (Phase 2a coexistence — additive, ignored by the
# current frontend). readiness_status is a LineStatus value verbatim.
plan_type: Optional[str] = None
readiness_status: Optional[str] = None
readiness_items: Optional[list[ReadinessItemOut]] = None
class UploadResponse(BaseModel):
@ -230,7 +209,6 @@ class UploadResponse(BaseModel):
stats: dict
mapping_summary: dict
batch_id: str | None = None
readiness_stats: dict | None = None
def _build_reason(
@ -331,11 +309,7 @@ class DocStatusRequest(BaseModel):
def _to_record_out(
r,
record=None,
confirmed_visit_date=None,
saved_doc_statuses: dict | None = None,
readiness=None,
r, record=None, confirmed_visit_date=None, saved_doc_statuses: dict | None = None
) -> RecordOut:
flag_val = r.flag.value if hasattr(r.flag, "value") else str(r.flag)
@ -417,21 +391,6 @@ def _to_record_out(
csv_transfer_from=getattr(record, "csv_transfer_from", None)
if record
else None,
plan_type=getattr(record, "plan_type", None) if record else None,
readiness_status=readiness.line_status.value if readiness else None,
readiness_items=[
ReadinessItemOut(
doc_type=i.doc_type,
required_state=i.required_state.value,
input_state=i.input_state.value,
quality=i.quality.value,
value=i.value,
satisfied=i.satisfied,
)
for i in readiness.doc_items
]
if readiness
else None,
)
@ -534,52 +493,32 @@ async def upload_csv(
records, as_of=date.today(), confirmed_visits=confirmed_visits
)
# Build a lookup from a tuple key to original record for doc state computation.
# Build a lookup from a tuple key to original record for doc state computation
record_lookup = {}
for r in records:
key = (r.patient_id, r.shipment_date, r.device_type, r.order_number)
record_lookup[key] = r
# Readiness index (Phase 2a: additive, augments the legacy scoring —
# replaces nothing). Row-to-line membership comes from the dedup output
# itself, never re-derived from a row's own plan_type.
index = build_readiness_index(records, confirmed_visits)
out = []
used_lines = {}
for r in results:
verdict, line_key, merged = resolve_group(
index, r.patient_id, r.device_type, r.last_shipment_date
)
if verdict is not None and line_key is not None:
used_lines[line_key] = verdict
# Exact row match first (byte-identical legacy path). CoverageResult
# carries no order_number, so the exact lookup misses for CSVs with an
# order-number column; the dedup merge then supplies the record fields
# (latest-non-null policy — the same values the verdict graded from).
rec = record_lookup.get(
out = [
_to_record_out(
r,
record=record_lookup.get(
(
r.patient_id,
r.last_shipment_date,
r.device_type,
getattr(r, "order_number", None),
)
)
if rec is None and merged is not None:
rec = merged_to_record(merged)
out.append(
_to_record_out(
r,
record=rec,
),
confirmed_visit_date=confirmed_visits.get(
hashlib.sha256(r.patient_id.encode()).hexdigest()
),
saved_doc_statuses=doc_statuses.get(
hashlib.sha256(r.patient_id.encode()).hexdigest(), {}
),
readiness=verdict,
)
)
for r in results
]
# Use real user ID and db_conn where possible if we implement it, hardcoding
# IP as 0.0.0.0 for now unless req object available
@ -614,10 +553,6 @@ async def upload_csv(
stats=_compute_stats(out),
mapping_summary=mapping_summary,
batch_id=batch_id,
# Only lines reachable from rows in THIS response, so the stats always
# reconcile with the records beside them (a record calculate_batch
# skipped cannot leave a phantom line in the counts).
readiness_stats=readiness_stats(used_lines),
)
@ -730,9 +665,6 @@ class ConfirmVisitRequest(BaseModel):
csv_transfer_from: str | None = None
order_number: str | None = None
hcpcs: str | None = None
# Client-mapped plan type echoed from the upload — never guessed. None
# keeps the recompute honest: the verdict reads Plan Type Needed.
plan_type: str | None = None
@app.post("/api/confirm-visit")
@ -815,12 +747,8 @@ async def confirm_visit(
csv_transfer_from=body.csv_transfer_from,
order_number=body.order_number,
hcpcs=body.hcpcs,
# Same normalization the CSV path applies (lowercase + trim, never
# guessed) so RecordOut.plan_type is consistent across both paths.
plan_type=(body.plan_type or "").strip().lower() or None,
)
result = calculate_coverage(record, confirmed_visit_date=confirmed)
verdict = readiness_for_record(record, confirmed_visit_date=confirmed)
from core.supabase_client import get_client
@ -835,9 +763,7 @@ async def confirm_visit(
db_conn=db_conn,
)
return _to_record_out(
result, record=record, confirmed_visit_date=confirmed, readiness=verdict
)
return _to_record_out(result, record=record, confirmed_visit_date=confirmed)
@app.put("/api/doc-status")

14
python-backend/api/normalizer.py
View file

@ -48,13 +48,6 @@ HEADER_MAP: dict[str, list[str]] = {
"plan", "plan_name", "plan name", "payer_name", "payer name",
"primary_payer", "primary payer", "ins_name", "carrier",
],
# Client-mapped plan type — grading-critical, NEVER derived from the payer
# name. Bare "plan"/"plan_name" stay payer aliases (a "Plan" column in the
# wild is a payer name); stealing them would silently re-map existing CSVs.
"plan_type": [
"plan_type", "plan type", "plan category", "plan_category",
"coverage type", "coverage_type", "insurance_type", "insurance type",
],
"component": [
"component", "item_type", "component_type", "type", "supply_type",
],
@ -319,12 +312,6 @@ def normalize_csv(text: str) -> tuple[list[ShipmentRecord], list[str], dict]:
order_number = mapped.get("order_number") or None
hcpcs = mapped.get("hcpcs") or None
# Client-supplied plan type, carried verbatim (lowercased and trimmed
# only — canonicalization of synonyms is a later, separate step).
# Anything the readiness engine does not recognize grades as
# "Plan Type Needed"; it is never guessed from the payer name.
plan_type = (mapped.get("plan_type") or "").strip().lower() or None
records.append(ShipmentRecord(
patient_id=patient_id,
device_type=device_type,
@ -340,7 +327,6 @@ def normalize_csv(text: str) -> tuple[list[ShipmentRecord], list[str], dict]:
csv_transfer_from=csv_transfer_from,
order_number=order_number,
hcpcs=hcpcs,
plan_type=plan_type,
))
return records, skipped, mapping_summary

203
python-backend/core/worklist_readiness.py
View file

@ -1,203 +0,0 @@
"""
worklist_readiness.py
Signal — STTIL Solutions
Orchestrates dedup + readiness for the live pipeline (Phase 2a coexistence,
docs/build-plans/01-readiness-model-wiring.md). The readiness verdict AUGMENTS
the coverage_calculator output: timing urgency (CoverageFlag) and documentation
completeness (LineStatus) are separate axes, combined at the worklist level.
This is the ONLY module that imports both dedup and readiness for the API.
core/coverage_calculator.py must never import this module (dedup.py imports
FROM coverage_calculator; the reverse edge would be circular).
LINE MEMBERSHIP IS DECIDED BY DEDUP, NEVER RE-DERIVED FROM A ROW. A row's own
plan_type can differ from its dedup group's MERGED plan_type (conflicting
groups route to None — never guessed), so resolving a row's verdict from its
own plan_type can attach the WRONG line's verdict (a reachable false green,
caught in independent review 2026-07-06). All row-level resolution goes
through the (patient_id, device_type, DOS-as-date) group key via
ReadinessIndex, which records the dedup output itself.
Non-gradeable lines (non-CGM devices) are never graded: they carry no verdict,
matching the locked design (preserved and displayed, not graded).
PHI CONTRACT: patient_id is the sole crosswalk key; it is hashed in any log
line (dedup._hash_pid pattern). No names, SSNs, DOBs, or contact fields.
"""
from __future__ import annotations
import hashlib
from dataclasses import dataclass, field
from datetime import date
from typing import Optional
from core.coverage_calculator import ShipmentRecord
from core.dedup import (
UNKNOWN_PLAN_TYPE,
MergedShipment,
_as_date,
assign_to_coverage_lines,
dedup,
)
from core.readiness import ReadinessVerdict, evaluate_readiness
# (patient_id, device_type, plan_type_id) — mirrors CoverageLineKey.
LineKey = tuple[str, str, str]
# (patient_id, device_type, DOS-as-date) — the locked dedup group key.
GroupKey = tuple[str, str, date]
@dataclass
class ReadinessIndex:
"""Dedup-derived resolution index for one upload batch.
by_line: verdicts per gradeable coverage line
line_by_group: dedup group -> the line that group's shipment belongs to
merged_by_group: dedup group -> its MergedShipment (display-field truth)
"""
by_line: dict[LineKey, ReadinessVerdict] = field(default_factory=dict)
line_by_group: dict[GroupKey, LineKey] = field(default_factory=dict)
merged_by_group: dict[GroupKey, MergedShipment] = field(default_factory=dict)
def _confirmed_for(
patient_id: str, confirmed_visits: dict[str, date]
) -> Optional[date]:
"""Look up a staff-confirmed visit date by hashed patient_id.
confirmed_visits is keyed by sha256(patient_id) hex digest, the same keying
coverage_calculator.calculate_batch uses.
"""
return confirmed_visits.get(
hashlib.sha256(patient_id.encode()).hexdigest()
)
def build_readiness_index(
records: list[ShipmentRecord],
confirmed_visits: Optional[dict[str, date]] = None,
) -> ReadinessIndex:
"""Dedup records into coverage lines, grade each gradeable line, and
record the group->line membership map so row-level resolution never
re-derives it.
Grades the CURRENT shipment only (latest DOS per line — locked rule);
prior shipments are collapsed history. Non-gradeable lines are skipped.
"""
confirmed_visits = confirmed_visits or {}
merged = dedup(records)
lines = assign_to_coverage_lines(merged)
index = ReadinessIndex()
for key, line in lines.items():
if not line.gradeable:
continue
cur = line.current_shipment
index.by_line[(key.patient_id, key.device_type, key.plan_type_id)] = (
evaluate_readiness(
line.plan_type if line.plan_type != UNKNOWN_PLAN_TYPE else None,
csv_swo_status=cur.csv_swo_status,
csv_visit_date=cur.csv_visit_date,
confirmed_visit_date=_confirmed_for(
key.patient_id, confirmed_visits
),
csv_pecos_verified=cur.csv_pecos_verified,
csv_pa_status=cur.csv_pa_status,
csv_diagnosis_on_file=cur.csv_diagnosis_on_file,
)
)
for m in merged:
gkey: GroupKey = (m.patient_id, m.device_type, m.shipment_date)
index.merged_by_group[gkey] = m
# Membership from the MERGED plan_type — the same rule
# assign_to_coverage_lines used, so this cannot diverge from it.
index.line_by_group[gkey] = (
m.patient_id,
m.device_type,
m.plan_type or UNKNOWN_PLAN_TYPE,
)
return index
def readiness_by_line(
records: list[ShipmentRecord],
confirmed_visits: Optional[dict[str, date]] = None,
) -> dict[LineKey, ReadinessVerdict]:
"""Line verdicts only (convenience wrapper over build_readiness_index)."""
return build_readiness_index(records, confirmed_visits).by_line
def resolve_group(
index: ReadinessIndex,
patient_id: str,
device_type: str,
shipment_date,
) -> tuple[Optional[ReadinessVerdict], Optional[LineKey], Optional[MergedShipment]]:
"""Resolve a result row to (verdict, line_key, merged_shipment) through the
dedup group key. The only sanctioned row-level resolution path."""
gkey: GroupKey = (patient_id, device_type, _as_date(shipment_date))
merged = index.merged_by_group.get(gkey)
line_key = index.line_by_group.get(gkey)
verdict = index.by_line.get(line_key) if line_key else None
return verdict, line_key, merged
def merged_to_record(m: MergedShipment) -> ShipmentRecord:
"""Synthetic ShipmentRecord from a dedup merge, used when the exact
row lookup misses (order-numbered CSVs: CoverageResult carries no
order_number). Field policy matches dedup exactly (latest-non-null doc
fields, collected order numbers), so the legacy doc_state and the
readiness verdict grade from the SAME values and cannot contradict.
order_number/hcpcs join the collected display lists — display-only fields.
"""
return ShipmentRecord(
patient_id=m.patient_id,
device_type=m.device_type,
shipment_date=m.shipment_date,
quantity=m.total_quantity,
payer=m.payer,
component=m.components[0] if m.components else "sensor",
csv_visit_date=m.csv_visit_date,
csv_swo_status=m.csv_swo_status,
csv_pecos_verified=m.csv_pecos_verified,
csv_pa_status=m.csv_pa_status,
csv_diagnosis_on_file=m.csv_diagnosis_on_file,
csv_transfer_from=m.csv_transfer_from,
order_number=", ".join(m.order_numbers) if m.order_numbers else None,
hcpcs=", ".join(m.hcpcs_codes) if m.hcpcs_codes else None,
plan_type=m.plan_type,
)
def readiness_for_record(
record: ShipmentRecord, confirmed_visit_date: Optional[date] = None
) -> ReadinessVerdict:
"""Grade a single record as its own line (the Confirm Visit recompute path)."""
return evaluate_readiness(
record.plan_type,
csv_swo_status=record.csv_swo_status,
csv_visit_date=record.csv_visit_date,
confirmed_visit_date=confirmed_visit_date,
csv_pecos_verified=record.csv_pecos_verified,
csv_pa_status=record.csv_pa_status,
csv_diagnosis_on_file=record.csv_diagnosis_on_file,
)
def readiness_stats(verdicts: dict[LineKey, ReadinessVerdict]) -> dict:
"""Line-level counts per LineStatus value, for measurable live verification.
Callers pass only the lines actually reachable from response rows, so the
stats always reconcile with the records in the same response.
"""
counts: dict[str, int] = {}
for verdict in verdicts.values():
label = verdict.line_status.value
counts[label] = counts.get(label, 0) + 1
counts["lines_total"] = len(verdicts)
return counts

596
tests/test_worklist_readiness.py
View file

@ -1,596 +0,0 @@
"""
Tests for the readiness-model wiring (build plan 01).
Engine behavior is covered by test_dedup.py / test_readiness.py; these tests
target the SEAMS the wiring adds: the orchestration module, the /api/upload
augmentation, the /api/confirm-visit recompute, and back-compat invariants.
PHI CONTRACT: fixtures use synthetic patient_ids only. No names, DOBs, SSNs.
"""
import hashlib
import io
import sys
from datetime import date, timedelta
from pathlib import Path
import pytest
sys.path.insert(0, str(Path(__file__).parent.parent / "python-backend"))
from core.coverage_calculator import ShipmentRecord, calculate_batch # noqa: E402
from core.worklist_readiness import ( # noqa: E402
build_readiness_index,
merged_to_record,
readiness_by_line,
readiness_for_record,
readiness_stats,
resolve_group,
)
TODAY = date.today()
def _record(**kwargs) -> ShipmentRecord:
defaults = dict(
patient_id="PT-1001",
device_type="dexcom_g7",
shipment_date=TODAY - timedelta(days=5),
quantity=3,
payer="Medicare Part B",
component="sensor",
)
defaults.update(kwargs)
return ShipmentRecord(**defaults)
def _hash(pid: str) -> str:
return hashlib.sha256(pid.encode()).hexdigest()
# --- orchestration module (unit level) --------------------------------------
def test_readiness_by_line_full_evidence_is_clear_to_ship():
rec = _record(
plan_type="medicare",
csv_swo_status="On File",
csv_visit_date=TODAY - timedelta(days=30),
csv_pecos_verified="Yes",
csv_diagnosis_on_file="Yes",
)
verdicts = readiness_by_line([rec])
key = ("PT-1001", "dexcom_g7", "medicare")
assert key in verdicts
assert verdicts[key].line_status.value == "Clear to Ship"
def test_readiness_by_line_skips_non_gradeable_lines():
rec = _record(device_type="omnipod_5", plan_type="medicare")
verdicts = readiness_by_line([rec])
assert verdicts == {}
def test_no_plan_type_reads_plan_type_needed():
rec = _record(
csv_swo_status="On File",
csv_visit_date=TODAY - timedelta(days=30),
csv_pecos_verified="Yes",
csv_diagnosis_on_file="Yes",
)
verdicts = readiness_by_line([rec])
key = ("PT-1001", "dexcom_g7", "unknown")
assert verdicts[key].line_status.value == "Plan Type Needed"
by_type = {i.doc_type: i for i in verdicts[key].doc_items}
assert by_type["pecos"].required_state.value == "NOT_EVALUATED"
assert by_type["pa"].required_state.value == "NOT_EVALUATED"
def test_same_dos_different_order_numbers_is_one_line_verdict():
a = _record(order_number="ORD-1", plan_type="medicare")
b = _record(order_number="ORD-2", plan_type="medicare")
verdicts = readiness_by_line([a, b])
assert len(verdicts) == 1
def test_verdict_graded_from_latest_dos():
older = _record(
shipment_date=TODAY - timedelta(days=40),
plan_type="medicare",
csv_swo_status="On File",
csv_visit_date=TODAY - timedelta(days=45),
csv_pecos_verified="Yes",
csv_diagnosis_on_file="Yes",
)
# Latest DOS is missing PECOS -> the line verdict must reflect THIS row.
newer = _record(
shipment_date=TODAY - timedelta(days=5),
plan_type="medicare",
csv_swo_status="On File",
csv_visit_date=TODAY - timedelta(days=45),
csv_diagnosis_on_file="Yes",
)
verdicts = readiness_by_line([older, newer])
assert len(verdicts) == 1
verdict = verdicts[("PT-1001", "dexcom_g7", "medicare")]
assert verdict.line_status.value == "Action Needed"
def test_conflicting_plan_types_route_to_unknown_line():
a = _record(order_number="ORD-1", plan_type="medicare")
b = _record(order_number="ORD-2", plan_type="medicaid")
index = build_readiness_index([a, b])
assert set(index.by_line) == {("PT-1001", "dexcom_g7", "unknown")}
verdict, line_key, merged = resolve_group(
index, "PT-1001", "dexcom_g7", a.shipment_date
)
assert verdict is not None
assert verdict.line_status.value == "Plan Type Needed"
assert line_key == ("PT-1001", "dexcom_g7", "unknown")
assert merged is not None
def test_mixed_plan_population_resolves_to_merged_line():
"""Review finding (HIGH family): a group where only SOME rows carry
plan_type merges to the typed line; row resolution must follow the MERGE,
not any single row's own plan_type."""
blank = _record(
order_number="ORD-1",
csv_swo_status="On File",
csv_visit_date=TODAY - timedelta(days=30),
csv_pecos_verified="Yes",
csv_diagnosis_on_file="Yes",
)
typed = _record(
order_number="ORD-2",
plan_type="medicare",
csv_swo_status="On File",
csv_visit_date=TODAY - timedelta(days=30),
csv_pecos_verified="Yes",
csv_diagnosis_on_file="Yes",
)
index = build_readiness_index([blank, typed])
assert set(index.by_line) == {("PT-1001", "dexcom_g7", "medicare")}
verdict, line_key, _ = resolve_group(
index, "PT-1001", "dexcom_g7", blank.shipment_date
)
assert verdict is not None, "blank-plan row must inherit its merged line's verdict"
assert verdict.line_status.value == "Clear to Ship"
assert line_key == ("PT-1001", "dexcom_g7", "medicare")
def test_conflicted_dos_never_borrows_another_lines_verdict():
"""Review finding (HIGH): a conflicted newer DOS must read its own
unknown-line verdict, never the older clean line's green."""
old_clean = _record(
shipment_date=TODAY - timedelta(days=45),
plan_type="medicare",
csv_swo_status="On File",
csv_visit_date=TODAY - timedelta(days=50),
csv_pecos_verified="Yes",
csv_diagnosis_on_file="Yes",
)
new_a = _record(order_number="ORD-1", plan_type="medicare")
new_b = _record(order_number="ORD-2", plan_type="medicaid")
index = build_readiness_index([old_clean, new_a, new_b])
assert set(index.by_line) == {
("PT-1001", "dexcom_g7", "medicare"),
("PT-1001", "dexcom_g7", "unknown"),
}
old_verdict, _, _ = resolve_group(
index, "PT-1001", "dexcom_g7", old_clean.shipment_date
)
new_verdict, _, _ = resolve_group(
index, "PT-1001", "dexcom_g7", new_a.shipment_date
)
assert old_verdict.line_status.value == "Clear to Ship"
assert new_verdict.line_status.value == "Plan Type Needed"
def test_merged_to_record_matches_dedup_field_policy():
"""Review finding (MEDIUM): fallback record fields must come from the
dedup merge (latest-non-null, collected order numbers), so the legacy
doc_state cannot contradict the readiness verdict."""
first = _record(order_number="ORD-1", csv_pecos_verified="Yes")
second = _record(order_number="ORD-2", plan_type="medicare")
index = build_readiness_index([first, second])
_, _, merged = resolve_group(
index, "PT-1001", "dexcom_g7", first.shipment_date
)
rec = merged_to_record(merged)
assert rec.csv_pecos_verified == "Yes" # latest-non-null keeps it
assert rec.order_number == "ORD-1, ORD-2" # collected, not arbitrary
assert rec.plan_type == "medicare"
def test_confirmed_visit_counts_as_positive_evidence():
rec = _record(
plan_type="medicare",
csv_swo_status="On File",
csv_pecos_verified="Yes",
csv_diagnosis_on_file="Yes",
)
confirmed = {_hash("PT-1001"): TODAY - timedelta(days=20)}
verdicts = readiness_by_line([rec], confirmed_visits=confirmed)
verdict = verdicts[("PT-1001", "dexcom_g7", "medicare")]
assert verdict.line_status.value == "Clear to Ship"
visit = next(i for i in verdict.doc_items if i.doc_type == "visit")
assert visit.satisfied
def test_readiness_stats_counts_lines():
good = _record(
patient_id="PT-A",
plan_type="medicare",
csv_swo_status="On File",
csv_visit_date=TODAY - timedelta(days=30),
csv_pecos_verified="Yes",
csv_diagnosis_on_file="Yes",
)
unknown = _record(patient_id="PT-B")
stats = readiness_stats(readiness_by_line([good, unknown]))
assert stats["Clear to Ship"] == 1
assert stats["Plan Type Needed"] == 1
assert stats["lines_total"] == 2
def test_no_raw_patient_id_in_logs(caplog):
# Conflicting plan types force the dedup warning path.
a = _record(patient_id="RAW-SECRET-ID-42", order_number="1", plan_type="medicare")
b = _record(patient_id="RAW-SECRET-ID-42", order_number="2", plan_type="medicaid")
with caplog.at_level("WARNING"):
readiness_by_line([a, b])
assert caplog.text # the warning fired
assert "RAW-SECRET-ID-42" not in caplog.text
def test_readiness_for_record_single_path():
rec = _record(
plan_type="medicare",
csv_swo_status="On File",
csv_pecos_verified="Yes",
csv_diagnosis_on_file="Yes",
)
verdict = readiness_for_record(
rec, confirmed_visit_date=TODAY - timedelta(days=10)
)
assert verdict.line_status.value == "Clear to Ship"
# --- API level (FastAPI TestClient) ------------------------------------------
httpx = pytest.importorskip("httpx", reason="httpx required for TestClient")
from fastapi.testclient import TestClient # noqa: E402
import api.main as api_main # noqa: E402
client = TestClient(api_main.app)
def _upload(csv_text: str):
return client.post(
"/api/upload",
files={"file": ("fixture.csv", io.BytesIO(csv_text.encode()), "text/csv")},
)
# Dates engineered so the legacy timing flag is RESUPPLY_READY (coverage ends
# within the 30-day window) and no renewal flag fires (visit recent).
SHIP = (TODAY - timedelta(days=5)).isoformat()
VISIT = (TODAY - timedelta(days=30)).isoformat()
HEADER = (
"patient_id,device_type,shipment_date,quantity,payer,plan_type,"
"swo_status,visit_date,pecos_verified,pa_status,diagnosis_on_file"
)
def test_upload_false_green_blocked_at_api_layer():
"""Acceptance case 1: legacy label green-ish, readiness verdict honest."""
csv_text = (
HEADER
+ f"\nPT-FG-1,Dexcom G7,{SHIP},3,Medicare Part B,medicare,"
+ f"On File,{VISIT},,,Yes"
)
resp = _upload(csv_text)
assert resp.status_code == 200
rec = resp.json()["records"][0]
assert rec["flag"] == "RESUPPLY_READY"
assert rec["status_label"] == "Clear to Ship" # legacy timing label
assert rec["readiness_status"] == "Action Needed" # PECOS absent blocks green
def test_upload_without_plan_type_reads_plan_type_needed():
"""Acceptance case 2: today's default for every record — honest, never guessed."""
header = (
"patient_id,device_type,shipment_date,quantity,payer,"
"swo_status,visit_date,pecos_verified,pa_status,diagnosis_on_file"
)
csv_text = (
header
+ f"\nPT-PTN-1,Dexcom G7,{SHIP},3,Medicare Part B,"
+ f"On File,{VISIT},Yes,,Yes"
)
resp = _upload(csv_text)
assert resp.status_code == 200
body = resp.json()
graded = [r for r in body["records"] if r["readiness_status"] is not None]
assert graded, "expected at least one graded record"
assert all(r["readiness_status"] == "Plan Type Needed" for r in graded)
assert body["readiness_stats"]["Plan Type Needed"] == body[
"readiness_stats"
]["lines_total"]
def test_upload_clear_to_ship_reachable_end_to_end():
"""Acceptance case 3: full positive evidence + mapped plan type = true green."""
csv_text = (
HEADER
+ f"\nPT-GREEN-1,Dexcom G7,{SHIP},3,Medicare Part B,medicare,"
+ f"On File,{VISIT},Yes,,Yes"
)
resp = _upload(csv_text)
assert resp.status_code == 200
rec = resp.json()["records"][0]
assert rec["plan_type"] == "medicare"
assert rec["readiness_status"] == "Clear to Ship"
items = {i["doc_type"]: i for i in rec["readiness_items"]}
assert items["pa"]["required_state"] == "NOT_REQUIRED" # Medicare FFS
assert all(i["satisfied"] for i in items.values())
def test_upload_with_order_numbers_still_resolves_records():
"""Regression for the record_lookup miss: CoverageResult has no
order_number, so exact-key lookups miss for order-numbered CSVs; the
fallback must recover doc_state AND readiness fields."""
header = HEADER + ",order_number"
csv_text = (
header
+ f"\nPT-ORD-1,Dexcom G7,{SHIP},3,Medicare Part B,medicare,"
+ f"On File,{VISIT},Yes,,Yes,ORD-778899"
)
resp = _upload(csv_text)
assert resp.status_code == 200
rec = resp.json()["records"][0]
assert rec["order_number"] == "ORD-778899"
assert rec["doc_state"] is not None
assert rec["readiness_status"] == "Clear to Ship"
def test_upload_backcompat_legacy_fields_unchanged():
"""Acceptance case 6: the wiring is additive — every legacy field matches
an independent reconstruction from the engines for an order-less CSV
(where the exact record lookup must hit and byte-identity must hold)."""
csv_text = (
HEADER
+ f"\nPT-BC-1,Dexcom G7,{SHIP},3,Medicare Part B,medicare,"
+ f"On File,{VISIT},Yes,,Yes"
+ f"\nPT-BC-2,Libre 3,{(TODAY - timedelta(days=90)).isoformat()},1,"
+ "Aetna,commercial,Pending,,,Requested,Yes"
)
resp = _upload(csv_text)
assert resp.status_code == 200
body = resp.json()
from api.normalizer import normalize_csv
from core.doc_state_machine import compute_doc_state
records, _, _ = normalize_csv(csv_text)
by_pid = {r.patient_id: r for r in records}
expected = calculate_batch(records, as_of=TODAY, confirmed_visits={})
assert len(body["records"]) == len(expected)
for got, want in zip(body["records"], expected):
rec = by_pid[want.patient_id]
assert got["patient_id"] == want.patient_id
assert got["flag"] == want.flag.value
assert got["priority_score"] == want.priority_score
assert got["status_label"] == api_main.FLAG_LABELS.get(
want.flag.value, want.flag.value
)
assert got["payer"] == want.payer
assert got["days_until_coverage_end"] == want.days_until_coverage_end
assert got["coverage_end_date"] == want.coverage_end_date.isoformat()
assert got["rule_version"] == want.rule_version
assert got["action"] == api_main.FLAG_ACTIONS.get(
want.flag.value, "Review"
)
# Record-echo fields resolve through the exact lookup (order-less CSV)
assert got["quantity"] == rec.quantity
assert got["order_number"] == rec.order_number
assert got["csv_swo_status"] == rec.csv_swo_status
assert got["csv_pecos_verified"] == rec.csv_pecos_verified
assert got["csv_pa_status"] == rec.csv_pa_status
assert got["csv_diagnosis_on_file"] == rec.csv_diagnosis_on_file
# doc_state reconstructed independently from the same inputs
doc = compute_doc_state(
payer_type=api_main._normalize_payer_type(want.payer),
csv_swo_status=rec.csv_swo_status,
csv_visit_date=rec.csv_visit_date,
confirmed_visit_date=None,
csv_pecos_verified=rec.csv_pecos_verified,
csv_pa_status=rec.csv_pa_status,
csv_diagnosis_on_file=rec.csv_diagnosis_on_file,
is_transfer=bool(rec.csv_transfer_from),
)
assert got["doc_state"]["visit"] == doc.visit
assert got["doc_state"]["pecos"] == doc.pecos
assert got["doc_state"]["diagnosis"] == doc.diagnosis
assert got["cascade"] == doc.cascade
# Full legacy stats dict reconstructed independently from the flags
flags = [w.flag.value for w in expected]
assert body["stats"] == {
"total": len(expected),
"supply_lapsed": flags.count("SUPPLY_LAPSED"),
"visit_required": flags.count("VISIT_REQUIRED"),
"transfer_pending": flags.count("TRANSFER_PENDING"),
"renewal_critical": flags.count("RENEWAL_CRITICAL"),
"renewal_elevated": flags.count("RENEWAL_ELEVATED"),
"renewal_soon": flags.count("RENEWAL_SOON"),
"resupply_ready": flags.count("RESUPPLY_READY"),
"active": flags.count("ACTIVE"),
"no_recent_shipment": flags.count("NO_RECENT_SHIPMENT"),
"prescriber_action": (
flags.count("SUPPLY_LAPSED")
+ flags.count("VISIT_REQUIRED")
+ flags.count("RENEWAL_CRITICAL")
+ flags.count("RENEWAL_ELEVATED")
),
}
def test_upload_mixed_plan_group_both_rows_carry_merged_verdict():
"""Review finding (HIGH family), API level: duplicate order-numbered rows
where only one carries plan_type — both output rows must carry the merged
line's verdict, and stats must reconcile with the rows."""
header = HEADER + ",order_number"
csv_text = (
header
+ f"\nPT-MIX-1,Dexcom G7,{SHIP},3,Medicare Part B,,"
+ f"On File,{VISIT},Yes,,Yes,ORD-1"
+ f"\nPT-MIX-1,Dexcom G7,{SHIP},3,Medicare Part B,medicare,"
+ f"On File,{VISIT},Yes,,Yes,ORD-2"
)
resp = _upload(csv_text)
assert resp.status_code == 200
body = resp.json()
assert len(body["records"]) == 2
for rec in body["records"]:
assert rec["readiness_status"] == "Clear to Ship"
assert rec["plan_type"] == "medicare"
assert rec["order_number"] == "ORD-1, ORD-2"
assert body["readiness_stats"] == {"Clear to Ship": 1, "lines_total": 1}
def test_upload_conflicted_dos_reads_own_line_not_older_green():
"""Review finding (HIGH), API level: the false-green repro. A conflicted
newer DOS must read Plan Type Needed even when an older clean DOS on the
same patient+device graded Clear to Ship."""
header = HEADER + ",order_number"
old_ship = (TODAY - timedelta(days=45)).isoformat()
old_visit = (TODAY - timedelta(days=50)).isoformat()
csv_text = (
header
+ f"\nPT-CONF-1,Dexcom G7,{old_ship},3,Medicare Part B,medicare,"
+ f"On File,{old_visit},Yes,,Yes,ORD-OLD"
+ f"\nPT-CONF-1,Dexcom G7,{SHIP},3,Medicare Part B,medicare,"
+ f"On File,{VISIT},Yes,,Yes,ORD-N1"
+ f"\nPT-CONF-1,Dexcom G7,{SHIP},3,Medicaid - GA,medicaid,"
+ f"On File,{VISIT},Yes,,Yes,ORD-N2"
)
resp = _upload(csv_text)
assert resp.status_code == 200
body = resp.json()
by_ship = {}
for rec in body["records"]:
by_ship.setdefault(rec["last_shipment_date"], []).append(rec)
for rec in by_ship[old_ship]:
assert rec["readiness_status"] == "Clear to Ship"
for rec in by_ship[SHIP]:
assert rec["readiness_status"] == "Plan Type Needed"
assert body["readiness_stats"] == {
"Clear to Ship": 1,
"Plan Type Needed": 1,
"lines_total": 2,
}
def test_upload_skipped_record_leaves_no_phantom_line_in_stats():
"""Review finding (LOW): a record calculate_batch skips (G7 transmitter
has no wear-day rule) must not leave a phantom line in readiness_stats."""
header = HEADER + ",component"
csv_text = (
header
+ f"\nPT-OK-1,Dexcom G7,{SHIP},3,Medicare Part B,medicare,"
+ f"On File,{VISIT},Yes,,Yes,sensor"
+ f"\nPT-TX-1,Dexcom G7,{SHIP},1,Medicare Part B,medicare,"
+ f"On File,{VISIT},Yes,,Yes,transmitter"
)
resp = _upload(csv_text)
assert resp.status_code == 200
body = resp.json()
pids = {r["patient_id"] for r in body["records"]}
assert pids == {"PT-OK-1"} # transmitter row skipped by the engine
assert body["readiness_stats"]["lines_total"] == 1
def test_upload_non_gradeable_rows_carry_no_verdict():
csv_text = (
HEADER
+ f"\nPT-POD-1,Omnipod 5,{SHIP},1,Medicare Part B,medicare,"
+ f"On File,{VISIT},Yes,,Yes"
)
resp = _upload(csv_text)
assert resp.status_code == 200
rec = resp.json()["records"][0]
assert rec["readiness_status"] is None
assert rec["readiness_items"] is None
def test_confirm_visit_recompute_flips_to_clear(monkeypatch):
"""Acceptance case 5: confirming the visit satisfies the last open
required item in the same response."""
monkeypatch.setattr(api_main, "get_or_create_org", lambda **kw: "org-test")
monkeypatch.setattr(
api_main, "upsert_confirmed_visit", lambda *a, **kw: True
)
# Upload path first: everything good except visit evidence.
csv_text = (
HEADER
+ f"\nPT-CV-1,Dexcom G7,{SHIP},3,Medicare Part B,medicare,"
+ "On File,,Yes,,Yes"
)
up = _upload(csv_text)
assert up.status_code == 200
assert up.json()["records"][0]["readiness_status"] == "Action Needed"
resp = client.post(
"/api/confirm-visit",
json={
"patient_id": "PT-CV-1",
"confirmed_date": (TODAY - timedelta(days=15)).isoformat(),
"shipment_date": SHIP,
"payer": "Medicare Part B",
"device_type": "dexcom_g7",
"quantity": 3,
"csv_swo_status": "On File",
"csv_pecos_verified": "Yes",
"csv_diagnosis_on_file": "Yes",
"plan_type": "medicare",
},
)
assert resp.status_code == 200
rec = resp.json()
assert rec["readiness_status"] == "Clear to Ship"
visit = next(i for i in rec["readiness_items"] if i["doc_type"] == "visit")
assert visit["satisfied"]
def test_confirm_visit_normalizes_plan_type(monkeypatch):
"""Review finding (LOW): both paths must return the same normalized
plan_type representation."""
monkeypatch.setattr(api_main, "get_or_create_org", lambda **kw: "org-test")
monkeypatch.setattr(
api_main, "upsert_confirmed_visit", lambda *a, **kw: True
)
resp = client.post(
"/api/confirm-visit",
json={
"patient_id": "PT-NORM-1",
"confirmed_date": (TODAY - timedelta(days=15)).isoformat(),
"shipment_date": SHIP,
"payer": "Medicare Part B",
"device_type": "dexcom_g7",
"quantity": 3,
"csv_swo_status": "On File",
"csv_pecos_verified": "Yes",
"csv_diagnosis_on_file": "Yes",
"plan_type": " Medicare ",
},
)
assert resp.status_code == 200
rec = resp.json()
assert rec["plan_type"] == "medicare"
assert rec["readiness_status"] == "Clear to Ship"