Add readiness-model dedup + coverage-line grouping foundation

Group normalized shipment rows into deduped shipments and gradeable
coverage lines (patient -> device -> shipment), the structural fix for the
row-keyed worklist bug (six months of orders no longer = six worklist lines).

- Locked dedup key (patient_id, device_type, date_of_service); order_number
  is display-only. Monthly resupplies stay separate, not duplicates.
- Client-mapped plan_type carried through, never guessed from payer name;
  conflicting plan types in a group route to "plan type needed".
- Quantity merge: sum across distinct component configs, max within a config
  (duplicate guard) with a warning breadcrumb on disagreement.
- CGM gradeable this phase; pumps preserved/displayed but not graded.

Additive only: not yet wired into the live pipeline, grading unchanged.
30 tests green (full suite 46), code-auditor reviewed (M1/M2/L1/L2/L3 fixed).

Design: docs/dedup-design-2026-06-24.md, docs/readiness-model-brief-2026-06-23.md

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

docs/ship-status-ledger.md

@@ -18,7 +18,7 @@ Canonical, git-tracked record of **what shipped, what didn't, and the reason or
 | 6 | Convex-Signal spike **design** | Pi | **SHIPPED** | docs/convex-signal-spike-design-2026-06-24.md. 40h/3-phase plan; Module map + Repository interface + Convex schema + Auth seam (Convex default, Clerk swappable) + Phase 1 (core readiness loop) = what Claude builds first. | Claude builds Phase 1 from it |
 | 7 | `signal-convex` repo creation | Claude (GitHub) + Kisa/token (Forgejo) | **PENDING** | Dependency: Pi design + Kisa approval (create once, correctly). Claude can do GitHub via `gh`; Forgejo has no API token here. | after design approved |
 | 8 | Convex BAA email (plan/cost/signing) | Kisa | **IN PROGRESS** | Kisa handling | Kisa sends |
-| 9 | Readiness-model build (rename coverage→readiness, rules-to-config, grouping+verdict, device override, require-plan-type, frontend nesting) | Claude | **UNBLOCKED** | Dedup decision locked: (patient_id, device_type, DOS). Order_number is display-only. Scales to all DMEPOS categories. Design: `docs/dedup-design-2026-06-24.md`. | Claude builds from dedup design + readiness model brief |
+| 9 | Readiness-model build (rename coverage→readiness, rules-to-config, grouping+verdict, device override, require-plan-type, frontend nesting) | Claude | **IN PROGRESS** | Step 3 grouping half SHIPPED 2026-06-26: `core/dedup.py` (CSVRow→MergedShipment→CoverageLine) + 30 tests green, code-auditor reviewed (no High; M1/M2/L1/L2/L3 fixed). Added additive `plan_type` field to ShipmentRecord (default None; live grading untouched). NOT yet wired into the live pipeline. Remaining: required-vs-input verdict engine, rules-to-addressable-config (IDs), device-keyed override+recompute+rollup, require-plan-type enforcement + remove name-guesser/default, frontend nesting + "plan type needed" state. | Next: build the verdict engine + rules-to-config; WIRE orchestration in `api/main.py` or a new pipeline module, NOT inside `coverage_calculator` (dedup imports from it → circular import). Each step code-auditor-reviewed before ship. |
 | 10 | NocoDB leads / CRM cleanup | Claude + Kisa | **BLOCKED** | Reason: leads table has no primary key (Signal-base data source `bdmhi02nks1pg6g` is corrupted — a fresh table there also came up PK-less; content base is healthy). SQLite can't add a PK to a populated table (Route 1 failed). Interim: capture still works (inserts via n8n `signal-cgm-lead`); the 2 test rows can't be deleted. | resolved by the data-platform decision (item 11) |
 | 11 | Data-platform decision (leads + future PHI backend) | Kisa (with Claude/Pi analysis) | **OPEN** | Options analyzed: repaired-NocoDB / Supabase Postgres / Neon / AWS (already set up, BAA+RDS+S3 parked) / Convex ($25/mo incl. BAA). Leads = non-PHI (stay outside PHI boundary). Decision informed by the Convex spike. | run spike → decide |
 | 12 | Research-versioning Layer-1 (`research-autopush.sh` + Stop-hook entry) | Claude | **NOT STARTED** | Dependency: Pi design exists (`docs/research-versioning-design-2026-06-24.md`). Lower priority than product. | next infra slot |

python-backend/core/coverage_calculator.py

@@ -76,6 +76,10 @@ class ShipmentRecord:
     csv_transfer_from: Optional[str] = None   # prior supplier name if transfer
     order_number: Optional[str] = None        # pass-through — billing system order/claim number
     hcpcs: Optional[str] = None               # pass-through — HCPCS code from supplier CSV
+    # Client-mapped plan type — REQUIRED for grading, NEVER derived from the payer name.
+    # medicare | medicare_advantage | medicaid | commercial | None(=plan type needed).
+    # The `payer` field above is the raw display name only (locked 2026-06-23).
+    plan_type: Optional[str] = None
 
 
 @dataclass

python-backend/core/dedup.py

@@ -0,0 +1,250 @@
+"""
+dedup.py
+Signal — STTIL Solutions
+
+Groups normalized shipment rows into deduped shipments and gradeable coverage
+lines. This is the structural foundation of the readiness model: it turns a
+flat list of CSV-derived rows into the patient -> device -> shipment hierarchy
+the worklist needs, BEFORE any grading happens.
+
+Design: docs/dedup-design-2026-06-24.md (Pi, locked by Kisa 2026-06-24)
+        docs/readiness-model-brief-2026-06-23.md sections 3, 5, 7
+
+LOCKED DEDUP KEY: (patient_id, device_type, date_of_service)
+    order_number is a displayed attribute, NEVER part of the dedup key.
+    Monthly resupplies are separate dates of service, NOT duplicates.
+
+PLAN TYPE: carried through verbatim from the client-mapped field. It is NEVER
+    derived from the payer name. None means "plan type needed" -> keyed as
+    'unknown' so the line still forms but cannot be graded green later.
+
+PHI CONTRACT: operates only on patient_id, device_type, shipment_date,
+    quantity, payer (display), plan_type, component, order_number, hcpcs, and
+    the csv_* doc-status fields. No names, SSNs, DOBs, or contact fields.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import logging
+from collections import defaultdict
+from dataclasses import dataclass, field
+from datetime import date, datetime
+from typing import Optional
+
+from core.coverage_calculator import ShipmentRecord, CoverageFlag
+
+logger = logging.getLogger(__name__)
+
+
+def _hash_pid(patient_id: str) -> str:
+    """Short hashed patient_id for log lines — raw patient_id is never logged."""
+    return hashlib.sha256(patient_id.encode()).hexdigest()[:12]
+
+
+def _as_date(value) -> date:
+    """Coerce a datetime to a pure date so same-day timestamps group together."""
+    return value.date() if isinstance(value, datetime) else value
+
+# CGM device types graded this phase. Pumps and other DMEPOS categories are
+# preserved and displayed but not graded (gradeable=False). Mirrors the CGM
+# devices in payer_rules.json; moves into addressable config in a later step.
+CGM_DEVICE_TYPES = frozenset(
+    {"dexcom_g6", "dexcom_g7", "freestyle_libre_2", "freestyle_libre_3"}
+)
+
+UNKNOWN_PLAN_TYPE = "unknown"
+
+
+def is_cgm_device(device_type: str) -> bool:
+    """True only for CGM device types graded in the current phase."""
+    return device_type in CGM_DEVICE_TYPES
+
+
+@dataclass
+class MergedShipment:
+    """One deduped shipment: all rows sharing (patient_id, device_type, DOS)."""
+    patient_id: str
+    device_type: str
+    shipment_date: date
+    total_quantity: int
+    components: list[str]                 # combined, deduped, first-seen order
+    order_numbers: list[str]              # collected, deduped — display only
+    hcpcs_codes: list[str]                # collected, deduped — display only
+    payer: str                           # raw display NAME, latest non-null
+    plan_type: Optional[str]             # client-mapped; NEVER from payer name
+    # csv_* doc fields — latest non-null wins across the group
+    csv_visit_date: Optional[date] = None
+    csv_swo_status: Optional[str] = None
+    csv_pecos_verified: Optional[str] = None
+    csv_pa_status: Optional[str] = None
+    csv_diagnosis_on_file: Optional[str] = None
+    csv_transfer_from: Optional[str] = None
+
+
+@dataclass(frozen=True)
+class CoverageLineKey:
+    patient_id: str
+    device_type: str
+    plan_type_id: str
+
+
+@dataclass
+class CoverageLine:
+    """A gradeable unit: one patient's one device under one plan type.
+
+    line_status and timing_flag are computed later by the grading step
+    (gradeLine); a freshly assigned line is intentionally ungraded.
+    """
+    patient_id: str
+    device_type: str
+    plan_type: str
+    gradeable: bool
+    current_shipment: MergedShipment
+    prior_shipments: list[MergedShipment] = field(default_factory=list)
+    line_status: Optional[str] = None
+    timing_flag: Optional[CoverageFlag] = None
+
+
+def _qty(record: ShipmentRecord) -> int:
+    """A row without a quantity is still a shipment -> default to 1."""
+    return 1 if record.quantity is None else record.quantity
+
+
+def _merge_quantities(rows: list[ShipmentRecord]) -> int:
+    """Sum quantities across DISTINCT component configs, max within a config.
+
+    Within one component config, identical rows are likely data-entry
+    duplicates, so take the MAX (guards against doubling). Across distinct
+    configs (a real component split, e.g. sensor + transmitter), SUM.
+
+    A disagreement within one config (e.g. qty 3 and qty 100 for the same
+    component) collapses to the max and leaves a warning, so silently dropping
+    quantity always leaves an audit breadcrumb.
+    """
+    by_config: dict[frozenset, list[int]] = defaultdict(list)
+    for row in rows:
+        by_config[frozenset({row.component})].append(_qty(row))
+
+    total = 0
+    for config, qtys in by_config.items():
+        if len(set(qtys)) > 1:
+            logger.warning(
+                "Quantity disagreement in dedup group patient=%s device=%s "
+                "component=%s qtys=%s -> taking max",
+                _hash_pid(rows[0].patient_id), rows[0].device_type,
+                set(config), qtys,
+            )
+        total += max(qtys)
+    return total
+
+
+def _merge_plan_type(rows: list[ShipmentRecord]) -> Optional[str]:
+    """Plan type is grading-critical and must NEVER be guessed.
+
+    If a dedup group carries more than one distinct non-null plan type, the data
+    is ambiguous, so return None ('plan type needed') rather than pick one.
+    """
+    distinct = {r.plan_type for r in rows if r.plan_type}
+    if len(distinct) > 1:
+        logger.warning(
+            "Conflicting plan_type in dedup group patient=%s device=%s: %s "
+            "-> routing to 'plan type needed'",
+            _hash_pid(rows[0].patient_id), rows[0].device_type, sorted(distinct),
+        )
+        return None
+    return next(iter(distinct)) if distinct else None
+
+
+def _dedup_preserve_order(values: list[Optional[str]]) -> list[str]:
+    """Collect non-empty values, dedup, preserve first-seen order."""
+    seen: dict[str, None] = {}
+    for v in values:
+        if v:
+            seen.setdefault(v, None)
+    return list(seen.keys())
+
+
+def _latest_non_null(rows: list[ShipmentRecord], attr: str):
+    """Last meaningful value for an attribute, in ingestion order.
+
+    Empty / whitespace-only strings are treated as null so a blank later row
+    never clobbers an earlier real value (does not rely on upstream nulling).
+    """
+    result = None
+    for row in rows:
+        value = getattr(row, attr)
+        if value is None:
+            continue
+        if isinstance(value, str) and value.strip() == "":
+            continue
+        result = value
+    return result
+
+
+def dedup(records: list[ShipmentRecord]) -> list[MergedShipment]:
+    """Group rows by (patient_id, device_type, shipment_date) into MergedShipments."""
+    groups: dict[tuple, list[ShipmentRecord]] = {}
+    for record in records:
+        key = (record.patient_id, record.device_type, _as_date(record.shipment_date))
+        groups.setdefault(key, []).append(record)
+
+    merged: list[MergedShipment] = []
+    for (patient_id, device_type, shipment_date), rows in groups.items():
+        merged.append(
+            MergedShipment(
+                patient_id=patient_id,
+                device_type=device_type,
+                shipment_date=shipment_date,
+                total_quantity=_merge_quantities(rows),
+                components=_dedup_preserve_order([r.component for r in rows]),
+                order_numbers=_dedup_preserve_order([r.order_number for r in rows]),
+                hcpcs_codes=_dedup_preserve_order([r.hcpcs for r in rows]),
+                payer=_latest_non_null(rows, "payer") or "",
+                plan_type=_merge_plan_type(rows),
+                csv_visit_date=_latest_non_null(rows, "csv_visit_date"),
+                csv_swo_status=_latest_non_null(rows, "csv_swo_status"),
+                csv_pecos_verified=_latest_non_null(rows, "csv_pecos_verified"),
+                csv_pa_status=_latest_non_null(rows, "csv_pa_status"),
+                csv_diagnosis_on_file=_latest_non_null(rows, "csv_diagnosis_on_file"),
+                csv_transfer_from=_latest_non_null(rows, "csv_transfer_from"),
+            )
+        )
+
+    merged.sort(key=lambda m: (m.patient_id, m.device_type, m.shipment_date))
+    return merged
+
+
+def assign_to_coverage_lines(
+    merged_shipments: list[MergedShipment],
+) -> dict[CoverageLineKey, CoverageLine]:
+    """Group MergedShipments into CoverageLines by (patient_id, device_type, plan_type).
+
+    plan_type is the client-mapped field. None is keyed as 'unknown' (plan type
+    needed) so the line still forms but cannot be graded green later. The latest
+    DOS in each line is the current shipment; earlier ones are collapsed history.
+    """
+    by_key: dict[CoverageLineKey, list[MergedShipment]] = {}
+    for shipment in merged_shipments:
+        key = CoverageLineKey(
+            patient_id=shipment.patient_id,
+            device_type=shipment.device_type,
+            plan_type_id=shipment.plan_type or UNKNOWN_PLAN_TYPE,
+        )
+        by_key.setdefault(key, []).append(shipment)
+
+    lines: dict[CoverageLineKey, CoverageLine] = {}
+    for key, shipments in by_key.items():
+        shipments.sort(key=lambda s: s.shipment_date, reverse=True)
+        lines[key] = CoverageLine(
+            patient_id=key.patient_id,
+            device_type=key.device_type,
+            plan_type=key.plan_type_id,
+            gradeable=is_cgm_device(key.device_type),
+            current_shipment=shipments[0],
+            prior_shipments=shipments[1:],
+            line_status=None,
+            timing_flag=None,
+        )
+
+    return lines

tests/test_dedup.py

@@ -0,0 +1,347 @@
+"""
+Tests for the dedup / grouping foundation of the readiness model.
+
+Source design: docs/dedup-design-2026-06-24.md (Pi, locked by Kisa 2026-06-24)
+              docs/readiness-model-brief-2026-06-23.md sections 3, 5, 7
+
+Locked dedup key: (patient_id, device_type, date_of_service).
+order_number is a displayed attribute, NEVER part of the dedup key.
+"""
+
+import sys
+from pathlib import Path
+sys.path.insert(0, str(Path(__file__).parent.parent / "python-backend"))
+
+from datetime import date, timedelta
+
+from core.coverage_calculator import ShipmentRecord
+from core.dedup import (
+    dedup,
+    assign_to_coverage_lines,
+    MergedShipment,
+    CoverageLine,
+    is_cgm_device,
+)
+
+DOS = date(2026, 6, 1)
+
+
+def make_row(**kwargs):
+    defaults = dict(
+        patient_id="P1",
+        device_type="dexcom_g7",
+        shipment_date=DOS,
+        quantity=3,
+        payer="Medicare Part B",
+        component="sensor",
+        plan_type="medicare",
+    )
+    defaults.update(kwargs)
+    return ShipmentRecord(**defaults)
+
+
+# ---------------------------------------------------------------------------
+# Dedup: grouping by (patient_id, device_type, DOS)
+# ---------------------------------------------------------------------------
+
+def test_dedup_component_split():
+    """Two rows, same patient+device+DOS, different components -> one MergedShipment, summed qty."""
+    rows = [
+        make_row(quantity=3, component="sensor", order_number="ORD-100"),
+        make_row(quantity=1, component="transmitter", order_number="ORD-101"),
+    ]
+    result = dedup(rows)
+    assert len(result) == 1
+    assert result[0].total_quantity == 4
+    assert set(result[0].components) == {"sensor", "transmitter"}
+    assert len(result[0].order_numbers) == 2
+
+
+def test_dedup_identical_duplicate():
+    """Two identical rows (same component) -> max quantity, not sum (duplicate guard)."""
+    rows = [
+        make_row(quantity=3, component="sensor"),
+        make_row(quantity=3, component="sensor"),
+    ]
+    result = dedup(rows)
+    assert len(result) == 1
+    assert result[0].total_quantity == 3  # max within config, not 6
+
+
+def test_dedup_mixed_duplicate_and_split():
+    """Two duplicate sensor rows + one transmitter row -> max(sensor)+max(transmitter) = 4."""
+    rows = [
+        make_row(quantity=3, component="sensor"),
+        make_row(quantity=3, component="sensor"),
+        make_row(quantity=1, component="transmitter"),
+    ]
+    result = dedup(rows)
+    assert result[0].total_quantity == 4
+
+
+def test_dedup_zero_quantity_row_does_not_add():
+    """qty=3 sensor + qty=0 sensor -> 3 (zero row is informational)."""
+    rows = [
+        make_row(quantity=3, component="sensor"),
+        make_row(quantity=0, component="sensor"),
+    ]
+    result = dedup(rows)
+    assert result[0].total_quantity == 3
+
+
+def test_dedup_wildly_different_same_component_takes_max():
+    """qty=3 + qty=100 same component -> max 100 (likely error, not a real split)."""
+    rows = [
+        make_row(quantity=3, component="sensor"),
+        make_row(quantity=100, component="sensor"),
+    ]
+    result = dedup(rows)
+    assert result[0].total_quantity == 100
+
+
+def test_dedup_different_dos():
+    """Same patient+device, different DOS -> two separate shipments."""
+    rows = [
+        make_row(shipment_date=DOS, quantity=3),
+        make_row(shipment_date=DOS + timedelta(days=30), quantity=3),
+    ]
+    result = dedup(rows)
+    assert len(result) == 2
+
+
+def test_dedup_different_device():
+    """Same patient+DOS, different device -> two separate shipments."""
+    rows = [
+        make_row(device_type="dexcom_g7", component="sensor"),
+        make_row(device_type="omnipod_5", component="pod"),
+    ]
+    result = dedup(rows)
+    assert len(result) == 2
+
+
+def test_dedup_no_quantity_defaults_to_one():
+    """Missing quantity -> default to 1 (a row without qty is still a shipment)."""
+    rows = [make_row(quantity=None, component="sensor")]
+    result = dedup(rows)
+    assert result[0].total_quantity == 1
+
+
+def test_dedup_missing_order_number():
+    """No order_number -> order_numbers list is empty, no crash."""
+    rows = [make_row(order_number=None)]
+    result = dedup(rows)
+    assert result[0].order_numbers == []
+
+
+def test_dedup_order_numbers_collected_and_deduped():
+    """Repeated order numbers across rows are collected once, in first-seen order."""
+    rows = [
+        make_row(component="sensor", order_number="ORD-100"),
+        make_row(component="transmitter", order_number="ORD-100"),
+    ]
+    result = dedup(rows)
+    assert result[0].order_numbers == ["ORD-100"]
+
+
+def test_dedup_hcpcs_collected():
+    """HCPCS codes collected across rows in the group."""
+    rows = [
+        make_row(component="sensor", hcpcs="A4239"),
+        make_row(component="transmitter", hcpcs="E2103"),
+    ]
+    result = dedup(rows)
+    assert set(result[0].hcpcs_codes) == {"A4239", "E2103"}
+
+
+def test_dedup_payer_latest_non_null_wins():
+    """payer is display-only; latest non-null row wins."""
+    rows = [
+        make_row(component="sensor", payer="Old Payer"),
+        make_row(component="transmitter", payer="Medicare Part B"),
+    ]
+    result = dedup(rows)
+    assert result[0].payer == "Medicare Part B"
+
+
+def test_dedup_csv_field_latest_non_null_wins():
+    """csv_swo_status: a later non-null value wins over an earlier null."""
+    rows = [
+        make_row(component="sensor", csv_swo_status=None),
+        make_row(component="transmitter", csv_swo_status="On File"),
+    ]
+    result = dedup(rows)
+    assert result[0].csv_swo_status == "On File"
+
+
+def test_dedup_csv_field_earlier_non_null_kept_when_later_is_null():
+    """csv_swo_status: an earlier non-null is kept when the later row is null."""
+    rows = [
+        make_row(component="sensor", csv_swo_status="On File"),
+        make_row(component="transmitter", csv_swo_status=None),
+    ]
+    result = dedup(rows)
+    assert result[0].csv_swo_status == "On File"
+
+
+def test_dedup_single_row_passthrough():
+    """A single row produces a MergedShipment carrying its fields."""
+    rows = [make_row(quantity=2, component="sensor", order_number="ORD-1")]
+    result = dedup(rows)
+    assert len(result) == 1
+    assert isinstance(result[0], MergedShipment)
+    assert result[0].patient_id == "P1"
+    assert result[0].device_type == "dexcom_g7"
+    assert result[0].shipment_date == DOS
+    assert result[0].total_quantity == 2
+    assert result[0].plan_type == "medicare"
+
+
+def test_dedup_plan_type_never_derived_from_payer():
+    """plan_type carries the client-mapped value; None stays None even with a known payer name."""
+    rows = [make_row(payer="Aetna PPO", plan_type=None)]
+    result = dedup(rows)
+    assert result[0].plan_type is None  # never guessed from "Aetna"
+
+
+# ---------------------------------------------------------------------------
+# CoverageLine assignment: group MergedShipments into gradeable lines
+# ---------------------------------------------------------------------------
+
+def test_is_cgm_device():
+    assert is_cgm_device("dexcom_g7") is True
+    assert is_cgm_device("freestyle_libre_3") is True
+    assert is_cgm_device("omnipod_5") is False
+
+
+def test_assign_groups_by_patient_device_plan():
+    """One patient, one device, one plan -> one CoverageLine."""
+    merged = dedup([
+        make_row(shipment_date=DOS, quantity=3),
+        make_row(shipment_date=DOS + timedelta(days=30), quantity=3),
+    ])
+    lines = assign_to_coverage_lines(merged)
+    assert len(lines) == 1
+
+
+def test_assign_latest_dos_is_current():
+    """The latest DOS shipment is the current one; older ones are prior."""
+    merged = dedup([
+        make_row(shipment_date=DOS, quantity=3),
+        make_row(shipment_date=DOS + timedelta(days=30), quantity=3),
+    ])
+    line = list(assign_to_coverage_lines(merged).values())[0]
+    assert line.current_shipment.shipment_date == DOS + timedelta(days=30)
+    assert len(line.prior_shipments) == 1
+    assert line.prior_shipments[0].shipment_date == DOS
+
+
+def test_assign_cgm_line_is_gradeable():
+    merged = dedup([make_row(device_type="dexcom_g7", component="sensor")])
+    line = list(assign_to_coverage_lines(merged).values())[0]
+    assert line.gradeable is True
+
+
+def test_assign_pump_line_not_gradeable():
+    merged = dedup([make_row(device_type="omnipod_5", component="pod")])
+    line = list(assign_to_coverage_lines(merged).values())[0]
+    assert line.gradeable is False
+
+
+def test_assign_different_plan_type_separate_lines():
+    """Same patient+device, different mapped plan_type -> two CoverageLines (dual-eligible)."""
+    merged = dedup([
+        make_row(shipment_date=DOS, plan_type="medicare"),
+        make_row(shipment_date=DOS + timedelta(days=30), plan_type="medicaid"),
+    ])
+    lines = assign_to_coverage_lines(merged)
+    assert len(lines) == 2
+
+
+def test_assign_none_plan_type_groups_as_unknown():
+    """plan_type None is keyed as 'unknown', not guessed, and still forms a line."""
+    merged = dedup([make_row(plan_type=None)])
+    lines = assign_to_coverage_lines(merged)
+    assert len(lines) == 1
+    line = list(lines.values())[0]
+    assert line.plan_type == "unknown"
+
+
+def test_assign_starts_ungraded():
+    """A freshly assigned line has no status yet (gradeLine runs later)."""
+    merged = dedup([make_row()])
+    line = list(assign_to_coverage_lines(merged).values())[0]
+    assert line.line_status is None
+    assert line.timing_flag is None
+
+
+# ---------------------------------------------------------------------------
+# Hardening (code-auditor findings M1, L1, L3 + edge cases)
+# ---------------------------------------------------------------------------
+
+def test_dedup_conflicting_plan_type_routes_to_unknown():
+    """Two rows in one group with DIFFERENT non-null plan types -> never guess.
+
+    The plan type is grading-critical and must never be silently picked. When a
+    dedup group disagrees, route to None so the line keys as 'plan type needed'.
+    """
+    rows = [
+        make_row(component="sensor", plan_type="medicare"),
+        make_row(component="transmitter", plan_type="medicaid"),
+    ]
+    result = dedup(rows)
+    assert len(result) == 1
+    assert result[0].plan_type is None
+    line = list(assign_to_coverage_lines(result).values())[0]
+    assert line.plan_type == "unknown"
+
+
+def test_dedup_agreeing_plan_type_preserved():
+    """Same non-null plan type across rows is kept (no false conflict)."""
+    rows = [
+        make_row(component="sensor", plan_type="medicare"),
+        make_row(component="transmitter", plan_type="medicare"),
+    ]
+    result = dedup(rows)
+    assert result[0].plan_type == "medicare"
+
+
+def test_dedup_empty_string_csv_field_does_not_clobber():
+    """An empty-string csv value must not overwrite an earlier real value."""
+    rows = [
+        make_row(component="sensor", csv_swo_status="On File"),
+        make_row(component="transmitter", csv_swo_status=""),
+    ]
+    result = dedup(rows)
+    assert result[0].csv_swo_status == "On File"
+
+
+def test_dedup_datetime_dos_coerced_to_date():
+    """Same calendar day with different timestamps -> one group (timezone edge)."""
+    from datetime import datetime
+    rows = [
+        make_row(shipment_date=datetime(2026, 6, 1, 9, 0), component="sensor"),
+        make_row(shipment_date=datetime(2026, 6, 1, 17, 0), component="transmitter"),
+    ]
+    result = dedup(rows)
+    assert len(result) == 1
+    assert result[0].shipment_date == date(2026, 6, 1)
+
+
+def test_dedup_empty_input():
+    assert dedup([]) == []
+    assert assign_to_coverage_lines([]) == {}
+
+
+def test_cgm_device_types_exist_in_payer_rules():
+    """Drift guard: every gradeable CGM device must exist in payer_rules.json,
+    and the known pump must not be in the gradeable set."""
+    import json
+    rules_path = (
+        Path(__file__).parent.parent
+        / "python-backend" / "config" / "payer_rules.json"
+    )
+    devices = set(json.loads(rules_path.read_text())["devices"].keys())
+    from core.dedup import CGM_DEVICE_TYPES
+    missing = CGM_DEVICE_TYPES - devices
+    assert not missing, f"CGM devices missing from payer_rules.json: {missing}"
+    assert "omnipod_5" not in CGM_DEVICE_TYPES