feat: wire readiness model into live pipeline (build plan 01, Phase 2a)

Readiness verdicts now ride additively on /api/upload and /api/confirm-visit:
- core/worklist_readiness.py: ReadinessIndex — row-to-line membership resolved
  through the dedup group key (patient_id, device_type, DOS), never re-derived
  from a row's own plan_type (kills a reviewer-reproduced false green)
- RecordOut gains plan_type / readiness_status / readiness_items (additive,
  default None); UploadResponse gains readiness_stats (reconciles with rows)
- Minimal plan_type CSV mapping (client-supplied, lowercased+trimmed, never
  guessed) per the umbrella reactivation plan; full enforcement lands in P5
- Fixes pre-existing record_lookup miss for order-numbered CSVs; fallback
  fields now come from the dedup MergedShipment (latest-non-null, collected
  order numbers) so doc_state and the verdict grade from the same values
- confirm-visit normalizes plan_type identically to the CSV path

Verification: 132 tests green (108 baseline + 24 wiring/regression);
adversarial 3-lens review + refutation pass (7 findings confirmed, all fixed,
regression-locked); independent code-auditor verdict SHIP; Pi E2E 33/33.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

docs/build-plans/01-readiness-model-wiring.md

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

python-backend/api/main.py

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

python-backend/api/normalizer.py

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

python-backend/core/worklist_readiness.py

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

tests/test_worklist_readiness.py

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