merge: SP32 rendering & service-provider NPI extraction into main

backend/src/cyclone/cli.py

@@ -297,6 +297,97 @@ def validate_tax_id_cmd(tax_id: str, log_level: str) -> None:
     sys.exit(1)
 
 
+# ---------------------------------------------------------------------------
+# SP32: `cyclone backfill-rendering-npi` (Task 6)
+#
+# Re-parses on-disk 837p/835 files and patches up the typed NPI columns
+# (``Claim.rendering_provider_npi`` / ``Remittance.rendering_provider_npi``)
+# for rows that were ingested before the T4 writers took effect.
+# ---------------------------------------------------------------------------
+
+
+@main.command("backfill-rendering-npi")
+@click.option(
+    "--file", "files",
+    multiple=True,
+    type=click.Path(exists=True, dir_okay=False, path_type=Path),
+    help=(
+        "Path to a specific file to re-parse. May be passed multiple times. "
+        "Mutually informative with --input-dir (both are processed)."
+    ),
+)
+@click.option(
+    "--input-dir",
+    type=click.Path(file_okay=False, path_type=Path),
+    default=None,
+    help=(
+        "Directory to scan one level deep for *.txt / *.edi / *.x12 files. "
+        "Honors CYCLONE_BACKFILL_INPUT_DIR if --input-dir is not passed."
+    ),
+)
+@click.option(
+    "--type", "transaction_type",
+    type=click.Choice(["837p", "835"]),
+    default=None,
+    help=(
+        "Pin the parser to use. Without --type, each file's transaction "
+        "kind is sniffed (filename hint + ISA/ST prefix)."
+    ),
+)
+@click.option(
+    "--log-level",
+    default="INFO",
+    show_default=True,
+    type=click.Choice(["DEBUG", "INFO", "WARNING", "ERROR"]),
+)
+def backfill_rendering_npi(
+    files: tuple[Path, ...],
+    input_dir: Path | None,
+    transaction_type: str | None,
+    log_level: str,
+) -> None:
+    """Re-parse on-disk X12 files to populate the rendering NPI columns.
+
+    Idempotent: rows whose ``rendering_provider_npi`` is already non-NULL
+    are left untouched. After patching, runs :func:`cyclone.reconcile.run`
+    once across every 835 batch so the T5 scoring arm can fire
+    retroactively on the touched pairs.
+
+    Exit code: 0 on a successful run (zero populated rows is still exit 0).
+    """
+    from cyclone import db as _db
+    from cyclone.store.backfill import backfill_rendering_provider_npi
+
+    # SP18: re-run setup so per-command --log-level overrides the
+    # group default. ``setup_logging`` is idempotent.
+    setup_logging(level=log_level)
+
+    # The backup pattern: each CLI subcommand that touches the DB
+    # initializes it explicitly so ``python -m cyclone.cli backup list``
+    # works on a fresh machine with no prior app boot.
+    _db.init_db()
+
+    # --input-dir falls back to CYCLONE_BACKFILL_INPUT_DIR.
+    if input_dir is None and not files:
+        env_dir = os.environ.get("CYCLONE_BACKFILL_INPUT_DIR", "").strip()
+        if env_dir:
+            input_dir = Path(env_dir)
+
+    summary = backfill_rendering_provider_npi(
+        files=list(files) if files else None,
+        input_dir=input_dir,
+        transaction_type=transaction_type,
+    )
+
+    # One-line summary so operators can grep a log scrape quickly.
+    click.echo(
+        f"claims_updated={summary.claims_updated} "
+        f"remits_updated={summary.remits_updated} "
+        f"files_processed={summary.files_processed} "
+        f"files_skipped={summary.files_skipped}"
+    )
+
+
 if __name__ == "__main__":
     main()
 

backend/src/cyclone/db.py

@@ -249,6 +249,7 @@ class Claim(Base):
     service_date_to: Mapped[Optional[date]] = mapped_column(Date, nullable=True)
     charge_amount: Mapped[Decimal] = mapped_column(Numeric(12, 2), nullable=False, default=Decimal("0"))
     provider_npi: Mapped[Optional[str]] = mapped_column(String(16), nullable=True)
+    rendering_provider_npi: Mapped[Optional[str]] = mapped_column(String(16), nullable=True)
     payer_id: Mapped[Optional[str]] = mapped_column(String(32), nullable=True)
     state: Mapped[ClaimState] = mapped_column(
         Enum(ClaimState, native_enum=False), nullable=False, default=ClaimState.SUBMITTED
@@ -344,6 +345,7 @@ class Remittance(Base):
     claim_id: Mapped[Optional[str]] = mapped_column(
         String(64), ForeignKey("claims.id"), nullable=True
     )
+    rendering_provider_npi: Mapped[Optional[str]] = mapped_column(String(16), nullable=True)
     status_code: Mapped[str] = mapped_column(String(4), nullable=False)
     status_label: Mapped[Optional[str]] = mapped_column(String(32), nullable=True)
     total_charge: Mapped[Decimal] = mapped_column(Numeric(12, 2), nullable=False, default=Decimal("0"))

backend/src/cyclone/migrations/0019_add_rendering_and_service_provider_npis.sql

@@ -0,0 +1,7 @@
+-- version: 19
+-- SP32: render & service-provider NPI extraction.
+-- Nullable: existing rows stay NULL until backfill runs.
+-- No indexes (used for set-equality, not range queries; nullable).
+
+ALTER TABLE claims ADD COLUMN rendering_provider_npi TEXT;
+ALTER TABLE remittances ADD COLUMN rendering_provider_npi TEXT;

backend/src/cyclone/parsers/models.py

@@ -142,6 +142,7 @@ class ClaimOutput(_Base):
     subscriber: Subscriber
     payer: Payer
     claim: ClaimHeader
+    rendering_provider_npi: str | None = None  # NM1*82 NM109 (Loop 2420A)
     diagnoses: list[Diagnosis] = Field(default_factory=list)
     service_lines: list[ServiceLine] = Field(default_factory=list)
     validation: ValidationReport

backend/src/cyclone/parsers/models_835.py

@@ -160,6 +160,7 @@ class ClaimPayment(_Base):
     ref_benefit_plan: str | None = None  # REF*CE per the CO guide
     service_payments: list[ServicePayment] = Field(default_factory=list)
     raw_segments: list[list[str]] = Field(default_factory=list)
+    service_provider_npi: str | None = None  # NM1*1P NM109 (Loop 2100 service provider)
 
     @model_validator(mode="before")
     @classmethod

backend/src/cyclone/parsers/parse_835.py

@@ -376,6 +376,7 @@ def _consume_claim_payment(segments: list[list[str]], idx: int) -> tuple[ClaimPa
 
     service_payments: list[ServicePayment] = []
     ref_benefit_plan: str | None = None
+    service_provider_npi: str | None = None
     per_diem: Decimal | None = None
     status_label = claim_status_label(status)
 
@@ -422,9 +423,16 @@ def _consume_claim_payment(segments: list[list[str]], idx: int) -> tuple[ClaimPa
             except ValueError:
                 per_diem = None
         elif s[0] == "NM1":
-            # Patient (QC) / service-provider (1P) — captured in raw_segments.
+            # SP32: capture service-provider NPI from NM1*1P (Loop 2100).
-            # The 835 spec doesn't require a structured patient model in v1.
+            # NM108 (idx 8) carries the ID qualifier (typically "XX");
-            pass
+            # NM109 (idx 9) is the value. Some senders omit NM108; accept
+            # both forms but require a 10-digit ID.
+            if len(s) > 9 and s[1] == "1P":
+                if len(s) > 8 and s[8] == "XX" and s[9]:
+                    if s[9].isdigit() and len(s[9]) == 10:
+                        service_provider_npi = s[9]
+                elif s[9] and s[9].isdigit() and len(s[9]) == 10:
+                    service_provider_npi = s[9]
         elif s[0] == "DTM":
             # Claim-level dates — captured in raw_segments.
             pass
@@ -446,6 +454,7 @@ def _consume_claim_payment(segments: list[list[str]], idx: int) -> tuple[ClaimPa
             ref_benefit_plan=ref_benefit_plan,
             service_payments=service_payments,
             raw_segments=raw,
+            service_provider_npi=service_provider_npi,
         ),
         idx,
     )

backend/src/cyclone/parsers/parse_837.py

@@ -210,6 +210,7 @@ def _consume_claim(segments: list[list[str]], idx: int) -> tuple[ClaimOutput, in
     service_lines: list[ServiceLine] = []
     raw: list[list[str]] = [seg]
     prior_auth: str | None = None
+    rendering_provider_npi: str | None = None
 
     idx += 1
     while idx < len(segments) and segments[idx][0] not in {"HL", "CLM", "SE"}:
@@ -226,6 +227,11 @@ def _consume_claim(segments: list[list[str]], idx: int) -> tuple[ClaimOutput, in
                 for code in parts[1:]:
                     if code:
                         diagnoses.append(Diagnosis(code=code, qualifier=qualifier))
+        elif s[0] == "NM1" and len(s) > 1 and s[1] == "82":
+            # SP32: capture rendering provider NPI from Loop 2420A NM1*82.
+            # NM108 (idx 8) is the qualifier (typically "XX"), NM109 (idx 9) is the NPI.
+            if len(s) > 9 and s[8] == "XX" and len(s[9]) == 10 and s[9].isdigit():
+                rendering_provider_npi = s[9]
         elif s[0] == "LX":
             line_no = int(s[1]) if len(s) > 1 and s[1].isdigit() else len(service_lines) + 1
             # LX is just a separator — the actual service line data is in the next SV1.
@@ -255,6 +261,7 @@ def _consume_claim(segments: list[list[str]], idx: int) -> tuple[ClaimOutput, in
             subscriber=Subscriber(first_name="", last_name="", member_id="", address=Address(line1="", city="", state="", zip="")),
             payer=Payer(name="", id=""),
             claim=claim_header,
+            rendering_provider_npi=rendering_provider_npi,
             diagnoses=diagnoses,
             service_lines=service_lines,
             validation=ValidationReport(passed=True, errors=[], warnings=[]),
@@ -296,7 +303,7 @@ def _consume_service_line(segments: list[list[str]], idx: int, line_no: int) ->
     provider_ref: str | None = None
 
     idx += 1
-    while idx < len(segments) and segments[idx][0] not in {"LX", "HL", "CLM", "SE"}:
+    while idx < len(segments) and segments[idx][0] not in {"LX", "HL", "CLM", "SE", "NM1"}:
         s = segments[idx]
         raw.append(s)
         if s[0] == "DTP" and len(s) > 2 and s[1] == "472":

backend/src/cyclone/reconcile.py

@@ -151,13 +151,16 @@ def _content_keys_match(remit, claim) -> set[str]:
       - shim / dataclass test objects (planned names: total_charge_amount,
         total_charge, rendering_provider_npi), and
       - real SQLAlchemy ORM instances (Claim.charge_amount, Remittance.total_charge,
-        Claim.provider_npi).
+        Claim.provider_npi, Claim.rendering_provider_npi,
+        Remittance.rendering_provider_npi).
 
-    Production note: the 835 parser stores ``rendering_provider_npi`` and
+    Production note (SP32): the typed ``rendering_provider_npi`` column is
-    ``total_charge_amount`` on ``Remittance.raw_json`` (the ORM has no
+    the primary read for both sides (Claim = NM1*82, Remit = NM1*1P). Legacy
-    dedicated column), so the read order below is:
+    rows pre-0019 still carry the value in ``raw_json``; the read order is:
-      - NPI:  transient attribute → raw_json
+      - NPI:  typed column → raw_json (``service_provider_npi`` for remit,
-      - Charge: planned attribute name → real ORM column → raw_json
+        ``rendering_provider_npi`` for claim) → claim ``provider_npi``
+        (billing fallback for legacy 837p rows without NM1*82 extraction).
+      - Charge: planned attribute name → real ORM column → raw_json.
     The ``raw_json`` fallback is what makes this helper safe against a
     raw ``Remittance`` ORM instance from ``reconcile.run()``.
     """
@@ -190,12 +193,13 @@ def _content_keys_match(remit, claim) -> set[str]:
        abs(_remit_charge - _claim_charge) < CHARGE_TOLERANCE:
         matched.add("charge")
 
-    # NPI: prefer attribute, fall back to raw_json (where the 835 parser
+    # NPI: typed-column primary path (SP32), raw_json fallback (legacy rows).
-    # stores rendering_provider_npi). The Remittance ORM has no NPI column,
+    # Remit reads rendering_provider_npi (single value, D4); claim reads
-    # so the raw_json path is the production read for raw ORM instances —
+    # rendering_provider_npi first, then falls back to provider_npi (billing)
-    # it must NOT be removed.
+    # for legacy rows where the 837p parser hadn't yet extracted NM1*82.
     remit_npi = (
         (getattr(remit, "rendering_provider_npi", "") or "")
+        or ((getattr(remit, "raw_json", None) or {}).get("service_provider_npi", "") or "")
         or ((getattr(remit, "raw_json", None) or {}).get("rendering_provider_npi", "") or "")
     ).strip()
     claim_npi = (

backend/src/cyclone/store/backfill.py

@@ -0,0 +1,340 @@
+"""SP32 Task 6: backfill rendering/service-provider NPIs from on-disk files.
+
+The T4 writers populate ``Claim.rendering_provider_npi`` (from NM1*82 in
+837P Loop 2420A) and ``Remittance.rendering_provider_npi`` (from the
+NM1*1P service-provider segment in 835 Loop 2100). Rows ingested before
+T4 was wired (or ingested via a path that bypasses the writer — e.g. an
+ad-hoc ``store.add`` from a notebook) still have a NULL column.
+
+This module re-parses on-disk X12 files and patches up those columns on
+matching rows. It is **idempotent**: rows whose
+``rendering_provider_npi`` is already non-NULL are left untouched, and
+re-running the same file twice is a clean no-op.
+
+Public surface
+--------------
+
+* :func:`backfill_rendering_provider_npi` — entry point used by the CLI.
+* :class:`BackfillSummary` — counts dataclass echoed back as a one-line
+  summary by the CLI.
+
+Reconcile is run once at the end so the new typed NPI arm (T5) can fire
+retroactively across the open claim/remit pairs the backfill touched.
+"""
+
+from __future__ import annotations
+
+import logging
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Iterable
+
+from sqlalchemy import select
+
+from cyclone import db
+from cyclone.parsers.parse_835 import parse as parse_835
+from cyclone.parsers.parse_837 import parse as parse_837
+from cyclone.parsers.payer import PayerConfig, PayerConfig835
+
+log = logging.getLogger(__name__)
+
+
+# ---------------------------------------------------------------------------
+# 837P fixture uses NM1*82 → rendering_provider_npi.
+# 835   fixture uses NM1*1P → service_provider_npi (mapped onto the same
+#        Remittance.rendering_provider_npi column by T4 _remittance_835_row).
+# ---------------------------------------------------------------------------
+
+
+@dataclass
+class BackfillSummary:
+    """Counts emitted by ``backfill_rendering_provider_npi``.
+
+    The CLI echoes a one-line summary (`claims_updated=N remits_updated=N
+    …`) at the end of the run.
+    """
+
+    claims_updated: int = 0
+    remits_updated: int = 0
+    files_processed: int = 0
+    files_skipped: int = 0
+
+
+# ---------------------------------------------------------------------------
+# Parser wrappers — keep exceptions local so one bad file can't abort
+# the whole backfill run.
+# ---------------------------------------------------------------------------
+
+
+def _parse_837_file(path: Path) -> tuple[list, Path | None]:
+    """Re-parse a single 837P file. Returns ``(claims, broken_alias)``.
+
+    On success, ``claims`` is the list of parsed ``ClaimOutput`` rows
+    and ``broken_alias`` is None. On any failure, ``claims`` is empty
+    and ``broken_alias`` is the same ``path`` so the caller can log
+    which file was skipped.
+    """
+    try:
+        text = path.read_text()
+    except OSError as exc:
+        log.warning("backfill: cannot read %s: %s", path, exc)
+        return [], path
+    try:
+        result = parse_837(text, PayerConfig.co_medicaid(), input_file=str(path))
+    except Exception as exc:  # noqa: BLE001 — failure isolated per-file
+        log.warning("backfill: 837 parse failed for %s: %s", path, exc)
+        return [], path
+    return list(result.claims), None
+
+
+def _parse_835_file(path: Path) -> tuple[list, Path | None]:
+    """Re-parse a single 835 file. Returns ``(claims, broken_alias)``.
+
+    ``claims`` is the list of parsed ``ClaimPayment`` rows.
+    """
+    try:
+        text = path.read_text()
+    except OSError as exc:
+        log.warning("backfill: cannot read %s: %s", path, exc)
+        return [], path
+    try:
+        result = parse_835(text, PayerConfig835.co_medicaid_835(), input_file=str(path))
+    except Exception as exc:  # noqa: BLE001
+        log.warning("backfill: 835 parse failed for %s: %s", path, exc)
+        return [], path
+    return list(result.claims), None
+
+
+# ---------------------------------------------------------------------------
+# DB-patching helpers — keep the per-row update logic in one place.
+# ---------------------------------------------------------------------------
+
+
+def _patch_claim_rendering_npi(parsed_claims: list) -> int:
+    """Update ``Claim.rendering_provider_npi`` for any parsed 837 claim.
+
+    Only rows whose column is currently NULL are touched. Returns the
+    count of rows updated.
+    """
+    from cyclone.db import Claim, SessionLocal
+
+    updated = 0
+    with SessionLocal()() as session:
+        for parsed in parsed_claims:
+            npi = getattr(parsed, "rendering_provider_npi", None)
+            if not npi:
+                continue
+            row = session.get(Claim, parsed.claim_id)
+            if row is None:
+                continue
+            if row.rendering_provider_npi is not None:
+                # Already populated (e.g. by a prior backfill run, or by
+                # the T4 writer if the claim was ingested afterwards).
+                continue
+            row.rendering_provider_npi = npi
+            updated += 1
+        if updated:
+            session.commit()
+    return updated
+
+
+def _patch_remit_rendering_npi(parsed_claims: list) -> int:
+    """Update ``Remittance.rendering_provider_npi`` for any parsed 835 claim.
+
+    The 835 NM1*1P segment maps onto ``ClaimPayment.service_provider_npi``
+    (which the T4 writer copies into ``Remittance.rendering_provider_npi``).
+    We match on ``payer_claim_control_number``; only NULL columns are
+    touched. Returns the count of rows updated.
+    """
+    from cyclone.db import Remittance, SessionLocal
+
+    updated = 0
+    with SessionLocal()() as session:
+        for parsed in parsed_claims:
+            npi = getattr(parsed, "service_provider_npi", None)
+            if not npi:
+                continue
+            target_pcn = getattr(parsed, "payer_claim_control_number", None)
+            if not target_pcn:
+                continue
+            row = session.execute(
+                select(Remittance).where(
+                    Remittance.payer_claim_control_number == target_pcn,
+                    Remittance.rendering_provider_npi.is_(None),
+                )
+            ).scalars().first()
+            if row is None:
+                continue
+            row.rendering_provider_npi = npi
+            updated += 1
+        if updated:
+            session.commit()
+    return updated
+
+
+# ---------------------------------------------------------------------------
+# Reconcile pass — run once across every 835 batch so the T5 scoring arm
+# can fire retroactively on pairs the backfill touched.
+# ---------------------------------------------------------------------------
+
+
+def _run_reconcile_sweep() -> int:
+    """Run :func:`cyclone.reconcile.run` over every 835 batch. Returns count.
+
+    Failures are isolated per-batch so one bad batch can't abort the
+    sweep — the goal is "best-effort retroactive reconcile".
+    """
+    from sqlalchemy import select as _select
+
+    from cyclone import reconcile as _reconcile
+    from cyclone.db import Batch, SessionLocal
+
+    completed = 0
+    with SessionLocal()() as session:
+        batch_ids = [
+            row[0] for row in session.execute(
+                _select(Batch.id).where(Batch.kind == "835")
+            ).all()
+        ]
+    # Each batch gets its own session — ``reconcile.run`` does not commit,
+    # so an exception in one batch must not orphan a half-flushed
+    # transaction.
+    for bid in batch_ids:
+        try:
+            with SessionLocal()() as s:
+                _reconcile.run(s, bid)
+                s.commit()
+            completed += 1
+        except Exception as exc:  # noqa: BLE001
+            log.warning("backfill: reconcile failed for batch %s: %s", bid, exc)
+    return completed
+
+
+# ---------------------------------------------------------------------------
+# Public entry point.
+# ---------------------------------------------------------------------------
+
+
+# Supported ``--type`` flag values; also ``None`` means auto-detect.
+TransactionType = str | None
+
+
+def backfill_rendering_provider_npi(
+    *,
+    files: Iterable[Path] | None = None,
+    input_dir: Path | None = None,
+    transaction_type: TransactionType = None,
+) -> BackfillSummary:
+    """Re-parse on-disk 837p + 835 files and populate the typed NPI columns.
+
+    Args:
+        files: explicit list of file paths to re-parse.
+        input_dir: directory to scan for ``*.txt`` / ``*.edi`` files.
+            Files are scanned one level deep.
+        transaction_type: ``"837p"`` or ``"835"``. ``None`` auto-detects
+            by attempting the 837p parser first and falling back to
+            the 835 parser.
+
+    Returns:
+        :class:`BackfillSummary` with populated counts.
+
+    Idempotent: only writes columns that are currently NULL; re-running
+    on the same files is a clean no-op. After patching, runs
+    :func:`cyclone.reconcile.run` over every 835 batch so the T5
+    scoring arm can re-fire on the touched pairs.
+    """
+    summary = BackfillSummary()
+
+    # 1. Resolve the candidate file set.
+    candidates: list[Path] = []
+    if files:
+        candidates.extend(Path(f) for f in files)
+    if input_dir is not None:
+        for ext in ("*.txt", "*.edi", "*.835", "*.837", "*.x12"):
+            candidates.extend(input_dir.glob(ext))
+    candidates = [p for p in candidates if p.is_file()]
+
+    # 2. Re-parse each file and patch matching rows.
+    for path in candidates:
+        kind = transaction_type or _sniff_kind(path)
+        if kind == "837p":
+            parsed_claims, broken = _parse_837_file(path)
+            if broken is not None:
+                summary.files_skipped += 1
+                continue
+            summary.files_processed += 1
+            summary.claims_updated += _patch_claim_rendering_npi(parsed_claims)
+        elif kind == "835":
+            parsed_claims, broken = _parse_835_file(path)
+            if broken is not None:
+                summary.files_skipped += 1
+                continue
+            summary.files_processed += 1
+            summary.remits_updated += _patch_remit_rendering_npi(parsed_claims)
+        else:
+            # Auto-detect tried both parsers and both failed → skip.
+            summary.files_skipped += 1
+
+    # 3. Reconcile sweep — let the T5 NPI arm fire retroactively.
+    if summary.claims_updated or summary.remits_updated:
+        try:
+            _run_reconcile_sweep()
+        except Exception:  # noqa: BLE001 — sweep is best-effort
+            log.exception("backfill: reconcile sweep failed")
+
+    return summary
+
+
+def _sniff_kind(path: Path) -> TransactionType:
+    """Best-effort transaction-type sniff (content + filename).
+
+    Returns ``"837p"``, ``"835"``, or ``None`` if both parsers fail.
+    Used only when the caller didn't pin ``transaction_type`` explicitly
+    via the CLI.
+    """
+    try:
+        text = path.read_text()
+    except OSError:
+        return None
+    # Cheap filename hint.
+    name = path.name.lower()
+    if name.endswith(".835") or "835" in name:
+        return _try_both(path, text, prefer="835")
+    if name.endswith(".837") or "837" in name or "837p" in name:
+        return _try_both(path, text, prefer="837p")
+    # Content: ISA + ST. 837 starts with "ST*837", 835 starts with "ST*835".
+    if "ST*837*" in text:
+        return "837p"
+    if "ST*835*" in text:
+        return "835"
+    # Fallback: try 837p parser, then 835.
+    return _try_both(path, text, prefer="837p")
+
+
+def _try_both(path: Path, text: str, *, prefer: str) -> TransactionType:
+    """Try the preferred parser first, then the other; return first winner."""
+    if prefer == "837p":
+        try:
+            parse_837(text, PayerConfig.co_medicaid(), input_file=str(path))
+            return "837p"
+        except Exception:
+            try:
+                parse_835(text, PayerConfig835.co_medicaid_835(), input_file=str(path))
+                return "835"
+            except Exception:
+                return None
+    try:
+        parse_835(text, PayerConfig835.co_medicaid_835(), input_file=str(path))
+        return "835"
+    except Exception:
+        try:
+            parse_837(text, PayerConfig.co_medicaid(), input_file=str(path))
+            return "837p"
+        except Exception:
+            return None
+
+
+__all__ = [
+    "BackfillSummary",
+    "backfill_rendering_provider_npi",
+]

backend/src/cyclone/store/orm_builders.py

@@ -53,6 +53,10 @@ def _claim_837_row(claim: ClaimOutput, batch_id: str) -> Claim:
         service_date_to=d_to,
         charge_amount=Decimal(claim.claim.total_charge or 0),
         provider_npi=claim.billing_provider.npi,
+        # SP32: wire NM1*82 (Loop 2420A) rendering provider NPI from
+        # ClaimOutput (T3 parser extraction) into the typed ORM column
+        # (T1 migration 0019). Falls back to None if absent.
+        rendering_provider_npi=claim.rendering_provider_npi,
         payer_id=claim.payer.id,
         state=ClaimState.SUBMITTED,
         raw_json=json.loads(claim.model_dump_json()),
@@ -89,6 +93,10 @@ def _remittance_835_row(cp: ClaimPayment, batch_id: str) -> Remittance:
         total_paid=Decimal(cp.total_paid or 0),
         patient_responsibility=cp.patient_responsibility,
         adjustment_amount=adjustment,
+        # SP32: wire NM1*1P (Loop 2100) service provider NPI from
+        # ClaimPayment (T2 parser extraction) into the typed ORM column
+        # (T1 migration 0019). Falls back to None if absent.
+        rendering_provider_npi=cp.service_provider_npi,
         received_at=received_at,
         service_date=service_date,
         is_reversal=cp.status_code in ("21", "22"),

backend/tests/conftest.py

@@ -146,9 +146,8 @@ def make_claim(db_session):
         claim_id=None
 
     `state=None` defaults to ``ClaimState.SUBMITTED`` so existing tests
-    that omit it keep behaving. `rendering_provider_npi` is stored as
+    that omit it keep behaving. `rendering_provider_npi` is wired through
-    a transient attribute (no ORM column) for ``_content_keys_match``
+    the real ``Claim.rendering_provider_npi`` column (SP32 migration 0019).
-    to read via ``getattr``.
     """
     def _make(
         patient_control_number: str,
@@ -171,10 +170,8 @@ def make_claim(db_session):
             charge_amount=total_charge,
             state=state if state is not None else _CS.SUBMITTED,
             matched_remittance_id=matched_remittance_id,
+            rendering_provider_npi=rendering_provider_npi,
         )
-        # Transient attribute — read by reconcile._content_keys_match
-        # via getattr; the Claim ORM has no rendering_provider_npi column.
-        c.rendering_provider_npi = rendering_provider_npi
         db_session.add(c)
         db_session.flush()
         return c
@@ -191,9 +188,8 @@ def make_remit(db_session):
         service_date, remit_id=None
 
     `total_charge_amount` is mapped to ``Remittance.total_charge`` (real ORM
-    field). `rendering_provider_npi` is stored as a transient attribute for
+    field). `rendering_provider_npi` is wired through the real
-    ``_content_keys_match`` to read via ``getattr`` (Remittance has no NPI
+    ``Remittance.rendering_provider_npi`` column (SP32 migration 0019).
-    column; the parser reads it from raw_json in production).
     """
     def _make(
         payer_claim_control_number: str,
@@ -216,11 +212,8 @@ def make_remit(db_session):
             received_at=datetime.now(timezone.utc),
             service_date=service_date,
             is_reversal=False,
+            rendering_provider_npi=rendering_provider_npi,
         )
-        # Transient attribute — read by reconcile._content_keys_match via
-        # getattr. The Remittance ORM has no rendering_provider_npi column;
-        # the 835 parser reads it from raw_json in production.
-        r.rendering_provider_npi = rendering_provider_npi
         db_session.add(r)
         db_session.flush()
         return r

backend/tests/fixtures/co_medicaid_837p_with_renderer.txt

@@ -0,0 +1,61 @@
+ISA*00*          *00*          *ZZ*11525703       *ZZ*COMEDASSISTPROG*260617*1937*^*00501*991102984*1*P*:~
+GS*HC*11525703*COMEDASSISTPROG*20260617*193715*991102984*X*005010X222A1~
+ST*837*991102984*005010X222A1~
+BHT*0019*00*co-fixture-001*20260617*193715*CH~
+NM1*41*2*Dzinesco*****46*11525703~
+PER*IC*Tester*EM*tester@example.com~
+NM1*40*2*COLORADO MEDICAL ASSISTANCE PROGRAM*****46*COMEDASSISTPROG~
+HL*1**20*1~
+PRV*BI*PXC*251E00000X~
+NM1*85*2*TOC, Inc.*****XX*1881068062~
+N3*1100 East Main St*Suite A~
+N4*Montrose*CO*814014063~
+REF*EI*721587149~
+HL*2*1*22*0~
+SBR*P*18*******MC~
+NM1*IL*1*Balliache*Marianela****MI*P060946~
+N3*1811 PAVILION DR APT 303~
+N4*Montrose*CO*814016072~
+DMG*D8*19590223*F~
+NM1*PR*2*COHCPF*****PI*SKCO0~
+CLM*t991102984o1c1d*85.40***12:B:1*Y*A*Y*Y~
+REF*G1*3173~
+HI*ABK:R69~
+LX*1~
+SV1*HC:T1019:U2*42.70*UN*6.00***1~
+DTP*472*D8*20260602~
+REF*6R*t991102984v769804d~
+LX*2~
+SV1*HC:T1019:U2*42.70*UN*6.00***1~
+DTP*472*D8*20260603~
+REF*6R*t991102984v770058d~
+NM1*82*2*RENDERING PROVIDER*****XX*1234567893~
+HL*3*1*22*0~
+SBR*P*18*******MC~
+NM1*IL*1*Barella*Victoria****MI*H582447~
+N3*1900 Kellie DR~
+N4*Montrose*CO*814019524~
+DMG*D8*19570727*F~
+NM1*PR*2*COHCPF*****PI*SKCO0~
+CLM*t991102984o1c2d*155.76***12:B:1*Y*A*Y*Y~
+REF*G1*3173~
+HI*ABK:R69~
+LX*1~
+SV1*HC:S5130:U2*38.94*UN*6.00***1~
+DTP*472*D8*20260602~
+REF*6R*t991102984v769805d~
+LX*2~
+SV1*HC:S5130:U2*38.94*UN*6.00***1~
+DTP*472*D8*20260603~
+REF*6R*t991102984v770059d~
+LX*3~
+SV1*HC:S5130:U2*38.94*UN*6.00***1~
+DTP*472*D8*20260604~
+REF*6R*t991102984v770308d~
+LX*4~
+SV1*HC:S5130:U2*38.94*UN*6.00***1~
+DTP*472*D8*20260605~
+REF*6R*t991102984v770668d~
+SE*45*991102984~
+GE*1*991102984~
+IEA*1*991102984~

backend/tests/test_acks.py

@@ -51,7 +51,7 @@ def test_migration_0002_creates_acks_table():
 
 def test_migration_latest_idempotent_on_fresh_db():
     """Re-running the migration on the same DB must be a no-op (PRAGMA
-    user_version already at the latest version — currently 18 after
+    user_version already at the latest version — currently 19 after
     0004-0006 line_reconciliation, 0005 ta1_acks, SP9's 0007
     providers/payers/clearhouse, SP10's 0008 payer_rejected,
     SP11's 0009 audit_log, SP14's 0010 payer_rejected_acknowledged,
@@ -60,15 +60,16 @@ def test_migration_latest_idempotent_on_fresh_db():
     SP22's 0015 drop_claims_unique_constraint, SP27-Task 11's 0016
     claims.matched_remittance_id index, SP27-Task 17's 0017
     claim.patient_control_number backfill UPDATE, SP28's 0018
-    claim_acks join table)."""
+    claim_acks join table, SP32's 0019
+    rendering_provider_npi + service_provider_npi)."""
     with db.engine().begin() as c:
         v1 = c.exec_driver_sql("PRAGMA user_version").scalar() or 0
-    assert v1 == 18
+    assert v1 == 19
     # A second run should not raise and should not bump the version.
     db_migrate.run(db.engine())
     with db.engine().begin() as c:
         v2 = c.exec_driver_sql("PRAGMA user_version").scalar() or 0
-    assert v2 == 18
+    assert v2 == 19
 
 
 def test_add_ack_persists_row():

backend/tests/test_backfill_cli.py

@@ -0,0 +1,227 @@
+"""SP32 Task 6: end-to-end smoke tests for the ``backfill-rendering-npi`` CLI.
+
+The CLI re-parses on-disk 837p + 835 files and backfills the new typed
+``rendering_provider_npi`` columns on existing Claim / Remittance rows
+that were ingested before the T4 writers took effect. Idempotent: rows
+that already have a non-NULL rendering NPI are left untouched.
+
+Test strategy mirrors ``test_cli_backup.py``: in-process Click CliRunner
+invocation against a fresh per-test DB (autouse ``_auto_init_db``
+fixture), pre-seeded with Claim/Remittance rows whose
+``rendering_provider_npi`` is NULL, then assert the CLI populated them.
+"""
+
+from __future__ import annotations
+
+from datetime import datetime, timezone
+from decimal import Decimal
+
+import pytest
+from click.testing import CliRunner
+
+from cyclone import db
+from cyclone.cli import main
+
+BACKEND_DIR = "backend"
+_REPO_ROOT = "."
+
+
+# ---------------------------------------------------------------------------
+# Fixtures
+# ---------------------------------------------------------------------------
+
+
+@pytest.fixture
+def seed_837p_claim_without_renderer():
+    """Insert a Claim row whose ``rendering_provider_npi`` is NULL.
+
+    The id (``t991102984o1c1d``) matches the first claim in
+    ``co_medicaid_837p_with_renderer.txt`` — the fixture that does
+    contain a NM1*82 rendering provider (NPI ``1234567893``) attached
+    to that claim. Backfill should populate the column from the
+    re-parsed file.
+    """
+    from cyclone.db import Batch, Claim, ClaimState
+    with db.SessionLocal()() as s:
+        s.add(Batch(
+            id="seed-837p-batch",
+            kind="837p",
+            input_filename="seed.x12",
+            parsed_at=datetime.now(timezone.utc),
+        ))
+        s.add(Claim(
+            id="t991102984o1c1d",
+            batch_id="seed-837p-batch",
+            patient_control_number="t991102984o1c1d",
+            charge_amount=Decimal("85.40"),
+            state=ClaimState.SUBMITTED,
+            rendering_provider_npi=None,  # ← the column under test
+            provider_npi="1881068062",
+        ))
+        s.commit()
+
+
+@pytest.fixture
+def seed_835_remit_without_service_provider(tmp_path):
+    """Insert a Remittance whose ``rendering_provider_npi`` is NULL.
+
+    Writes a minimal 835 fixture to ``tmp_path`` containing an
+    NM1*1P service provider (NPI ``2222222222``) attached to the
+    only CLP segment (claim control number ``CLM001``). Returns
+    the path to the written file.
+    """
+    from cyclone.db import Batch, Remittance
+
+    fixture = tmp_path / "minimal_835_with_service_provider.txt"
+    fixture.write_text(
+        "ISA*00*          *00*          *ZZ*RECEIVERID     *ZZ*SENDERID       "
+        "*250101*1200*^*00501*000000001*0*P*:~"
+        "GS*HP*RECEIVER*SENDER*20250101*1200*1*X*005010X221A1~"
+        "ST*835*0001~"
+        "BPR*I*85.40*C*NON*****01*021000021*DA*123456*1511111"
+        "**01*031302955*DA*9876543~"
+        "TRN*1*1511111*1511111~"
+        "DTM*405*20250101~"
+        "N1*PR*COLORADO MEDICAID*XV*CO MEDICAID~"
+        "N3*PO BOX 1100*~"
+        "N4*DENVER*CO*80203~"
+        "REF*2U*12345~"
+        "N1*PE*ACME CLINIC*XX*1111111111~"
+        "LX*1~"
+        "CLP*CLM001*1*85.40*85.40*0.00*MC*CLM001*11*1~"
+        "CAS*PR*1*0.00~"
+        "NM1*1P*2*RENDERING PROVIDER*****XX*2222222222~"
+        "SVC*HC:99213*85.40*85.40**1~"
+        "DTM*472*20250101~"
+        "SE*18*0001~"
+        "GE*1*1~"
+        "IEA*1*000000001~"
+    )
+
+    with db.SessionLocal()() as s:
+        s.add(Batch(
+            id="seed-835-batch",
+            kind="835",
+            input_filename=fixture.name,
+            parsed_at=datetime.now(timezone.utc),
+        ))
+        s.add(Remittance(
+            id="CLM001",
+            batch_id="seed-835-batch",
+            payer_claim_control_number="CLM001",
+            claim_id=None,
+            status_code="1",
+            status_label="Primary payer forward",
+            total_charge=Decimal("85.40"),
+            total_paid=Decimal("85.40"),
+            patient_responsibility=Decimal("0"),
+            adjustment_amount=Decimal("0"),
+            received_at=datetime.now(timezone.utc),
+            is_reversal=False,
+            rendering_provider_npi=None,  # ← the column under test
+        ))
+        s.commit()
+
+    return fixture
+
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+
+def _run(args: list[str]) -> object:
+    """Invoke the CLI in-process via CliRunner.
+
+    CliRunner auto-captures stdout/stderr; ``catch_exceptions=False``
+    so the test fails loudly on any unexpected raise instead of a
+    swallowed traceback.
+    """
+    return CliRunner().invoke(main, args, catch_exceptions=False)
+
+
+# ---------------------------------------------------------------------------
+# Tests
+# ---------------------------------------------------------------------------
+
+
+def test_backfill_populates_837p_rendering_npi(seed_837p_claim_without_renderer):
+    """Re-parse the 837p fixture → Claim.rendering_provider_npi populated."""
+    fixture_path = "tests/fixtures/co_medicaid_837p_with_renderer.txt"
+
+    # Sanity: column starts NULL.
+    with db.SessionLocal()() as s:
+        from cyclone.db import Claim
+        row = s.get(Claim, "t991102984o1c1d")
+        assert row is not None
+        assert row.rendering_provider_npi is None
+
+    result = _run(["backfill-rendering-npi", "--file", fixture_path, "--type", "837p"])
+    assert result.exit_code == 0, result.output
+
+    with db.SessionLocal()() as s:
+        from cyclone.db import Claim
+        row = s.get(Claim, "t991102984o1c1d")
+        assert row is not None
+        # First claim in the fixture carries NM1*82 NPI ``1234567893``.
+        assert row.rendering_provider_npi == "1234567893"
+    # CLI prints the summary line with the populated counts.
+    assert "claims_updated=" in result.output
+    assert "remits_updated=" in result.output
+
+
+def test_backfill_populates_835_rendering_npi(seed_835_remit_without_service_provider):
+    """Re-parse an 835 with NM1*1P → Remittance.rendering_provider_npi populated."""
+    fixture_path = str(seed_835_remit_without_service_provider)
+
+    # Sanity: column starts NULL.
+    with db.SessionLocal()() as s:
+        from cyclone.db import Remittance
+        row = s.get(Remittance, "CLM001")
+        assert row is not None
+        assert row.rendering_provider_npi is None
+
+    result = _run(["backfill-rendering-npi", "--file", fixture_path, "--type", "835"])
+    assert result.exit_code == 0, result.output
+
+    with db.SessionLocal()() as s:
+        from cyclone.db import Remittance
+        row = s.get(Remittance, "CLM001")
+        assert row is not None
+        # Inline fixture's NM1*1P NM109 is ``2222222222``.
+        assert row.rendering_provider_npi == "2222222222"
+    assert "remits_updated=" in result.output
+
+
+def test_backfill_is_idempotent(seed_837p_claim_without_renderer):
+    """Second run is a no-op — already-populated columns stay."""
+    fixture_path = "tests/fixtures/co_medicaid_837p_with_renderer.txt"
+
+    # First run: populates the column.
+    first = _run(["backfill-rendering-npi", "--file", fixture_path, "--type", "837p"])
+    assert first.exit_code == 0, first.output
+
+    # Second run: still exits 0; column is unchanged; nothing breaks.
+    second = _run(["backfill-rendering-npi", "--file", fixture_path, "--type", "837p"])
+    assert second.exit_code == 0, second.output
+
+    with db.SessionLocal()() as s:
+        from cyclone.db import Claim
+        row = s.get(Claim, "t991102984o1c1d")
+        assert row.rendering_provider_npi == "1234567893"
+
+
+def test_backfill_help_lists_subcommand():
+    """Regression guard: the subcommand is wired into ``main``."""
+    result = _run(["--help"])
+    assert result.exit_code == 0
+    assert "backfill-rendering-npi" in result.output
+
+
+def test_backfill_with_no_files_is_noop():
+    """Running with no --file and no --input-dir is a clean no-op (exit 0)."""
+    result = _run(["backfill-rendering-npi"])
+    assert result.exit_code == 0, result.output
+    # Summary line still printed (zero counts).
+    assert "claims_updated=0" in result.output
+    assert "remits_updated=0" in result.output

backend/tests/test_db_migrate.py

@@ -124,14 +124,16 @@ def test_migration_latest_idempotent_on_fresh_db(tmp_path: Path) -> None:
     SP27 Task 17 bumped it to 17 with the patient_control_number
     backfill UPDATE (the migration runner now applies DML too).
     SP28 bumped it to 18 with the claim_acks join table.
+    SP32 bumped it to 19 with rendering_provider_npi +
+    service_provider_npi on claims and remittances.
     """
     engine = _fresh_engine(tmp_path)
     db_migrate.run(engine)
     v_after_first = _user_version(engine)
-    assert v_after_first == 18, f"expected head=18, got {v_after_first}"
+    assert v_after_first == 19, f"expected head=19, got {v_after_first}"
 
     db_migrate.run(engine)
-    assert _user_version(engine) == 18, "second run should not bump version"
+    assert _user_version(engine) == 19, "second run should not bump version"
 
 
 def test_drop_claims_unique_constraint_migration(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
@@ -157,7 +159,7 @@ def test_drop_claims_unique_constraint_migration(tmp_path: Path, monkeypatch: py
     engine = _fresh_engine(tmp_path)
 
     db_migrate.run(engine)
-    assert _user_version(engine) == 18, f"expected head=18, got {_user_version(engine)}"
+    assert _user_version(engine) == 19, f"expected head=19, got {_user_version(engine)}"
 
     # Two claims in one batch with the same patient_control_number
     # must be insertable. If 0015's table recreation re-introduced a

backend/tests/test_parse_835.py

@@ -178,3 +178,35 @@ def test_parse_service_payment_units_default_unit_type_to_un():
     # Default to UN when SVC04 missing but units present.
     assert s1.unit_type == "UN", s1.unit_type
     assert s2.unit_type == "UN", s2.unit_type
+
+
+def test_parse_835_extracts_service_provider_npi_from_nm1_1p():
+    """SP32: NM1*1P in Loop 2100 populates ClaimPayment.service_provider_npi."""
+    from cyclone.parsers.parse_835 import parse as parse_835
+
+    text = (
+        "ISA*00*          *00*          *ZZ*RECEIVERID     *ZZ*SENDERID       *250101*1200*^*00501*000000001*0*P*:~"
+        "GS*HP*RECEIVER*SENDER*20250101*1200*1*X*005010X221A1~"
+        "ST*835*0001~"
+        "BPR*I*85.40*C*NON*****01*021000021*DA*123456*1511111**01*031302955*DA*9876543~"
+        "TRN*1*1511111*1511111~"
+        "DTM*405*20250101~"
+        "N1*PR*COLORADO MEDICAID*XV*CO MEDICAID~"
+        "N3*PO BOX 1100*~"
+        "N4*DENVER*CO*80203~"
+        "REF*2U*12345~"
+        "PER*BL*SUPPORT*TE*8005551212~"
+        "N1*PE*ACME CLINIC*XX*1111111111~"
+        "LX*1~"
+        "CLP*CLM001*1*85.40*85.40*0.00*MC*CLM001*11*1~"
+        "CAS*PR*1*0.00~"
+        "NM1*1P*2*RENDERING PROVIDER*****XX*2222222222~"
+        "SVC*HC:99213*85.40*85.40**1~"
+        "DTM*472*20250101~"
+        "SE*18*0001~"
+        "GE*1*1~"
+        "IEA*1*000000001~"
+    )
+    result = parse_835(text, payer_config=None)  # type: ignore[arg-type]
+    assert len(result.claims) == 1
+    assert result.claims[0].service_provider_npi == "2222222222"

backend/tests/test_parse_837.py

@@ -68,3 +68,16 @@ def test_parse_uses_generic_config_when_requested():
     assert len(result.claims) == 1
     # No patient-loop rule on generic config
     assert result.claims[0].validation.passed is True
+
+
+def test_parse_837_extracts_rendering_provider_npi_from_nm1_82():
+    """SP32: NM1*82 (rendering provider) populates ClaimOutput.rendering_provider_npi."""
+    from pathlib import Path
+
+    text = Path("tests/fixtures/co_medicaid_837p_with_renderer.txt").read_text()
+    result = parse(text, PayerConfig.co_medicaid())
+    assert len(result.claims) >= 1
+    # Find the claim that has the rendering NPI (others may not).
+    matches = [c for c in result.claims if c.rendering_provider_npi is not None]
+    assert len(matches) == 1
+    assert matches[0].rendering_provider_npi == "1234567893"

backend/tests/test_reconcile.py

@@ -792,3 +792,104 @@ def test_reconcile_run_no_match_leaves_remit_unlinked(db_session, make_claim, ma
     assert claim.state == ClaimState.SUBMITTED  # unchanged
     db_session.refresh(remit)
     assert remit.claim_id is None  # unchanged
+
+
+# --- SP32 Task 4: ORM builders wire rendering_provider_npi -----------------
+
+
+def test_claim_837_row_includes_rendering_provider_npi():
+    """SP32: Claim.rendering_provider_npi is populated from ClaimOutput in ORM builder."""
+    from pathlib import Path
+    from cyclone.parsers.parse_837 import parse
+    from cyclone.parsers.payer import PayerConfig
+    from cyclone.store.orm_builders import _claim_837_row
+
+    text = Path("tests/fixtures/co_medicaid_837p_with_renderer.txt").read_text()
+    parsed = parse(text, PayerConfig.co_medicaid())
+    # Find the claim whose rendering NPI was extracted.
+    targets = [c for c in parsed.claims if c.rendering_provider_npi == "1234567893"]
+    assert len(targets) == 1
+    row = _claim_837_row(targets[0], batch_id="test-batch")
+    assert row.rendering_provider_npi == "1234567893"
+
+
+def test_remittance_835_row_includes_service_provider_npi():
+    """SP32: Remittance.rendering_provider_npi is populated from ClaimPayment in ORM builder."""
+    from cyclone.parsers.parse_835 import parse as parse_835
+    from cyclone.store.orm_builders import _remittance_835_row
+
+    text = (
+        "ISA*00*          *00*          *ZZ*RECEIVERID     *ZZ*SENDERID       *250101*1200*^*00501*000000001*0*P*:~"
+        "GS*HP*RECEIVER*SENDER*20250101*1200*1*X*005010X221A1~"
+        "ST*835*0001~"
+        "BPR*I*85.40*C*NON*****01*021000021*DA*123456*1511111**01*031302955*DA*9876543~"
+        "TRN*1*1511111*1511111~"
+        "DTM*405*20250101~"
+        "N1*PR*COLORADO MEDICAID*XV*CO MEDICAID~"
+        "N3*PO BOX 1100*~"
+        "N4*DENVER*CO*80203~"
+        "REF*2U*12345~"
+        "PER*BL*SUPPORT*TE*8005551212~"
+        "N1*PE*ACME CLINIC*XX*1111111111~"
+        "LX*1~"
+        "CLP*CLM001*1*85.40*85.40*0.00*MC*CLM001*11*1~"
+        "CAS*PR*1*0.00~"
+        "NM1*1P*2*RENDERING PROVIDER*****XX*2222222222~"
+        "SVC*HC:99213*85.40*85.40**1~"
+        "DTM*472*20250101~"
+        "SE*18*0001~"
+        "GE*1*1~"
+        "IEA*1*000000001~"
+    )
+    parsed = parse_835(text, payer_config=None)  # type: ignore[arg-type]
+    assert len(parsed.claims) == 1
+    assert parsed.claims[0].service_provider_npi == "2222222222"
+    row = _remittance_835_row(parsed.claims[0], batch_id="test-batch")
+    assert row.rendering_provider_npi == "2222222222"
+
+
+# --- SP32 Task 5: typed-column NPI preference in _content_keys_match -------
+
+
+def test_content_keys_match_typed_npi_columns_match(db_session, make_claim, make_remit):
+    """SP32: typed-column NPI contributes to the matched set when both sides have it."""
+    from cyclone.reconcile import _content_keys_match
+    from datetime import date
+    claim = make_claim(
+        patient_control_number="PCN1",
+        total_charge=Decimal("85.40"),
+        rendering_provider_npi="2222222222",
+        service_date_from=date(2025, 1, 1),
+    )
+    remit = make_remit(
+        payer_claim_control_number="PCN1",
+        total_charge_amount=Decimal("85.40"),
+        rendering_provider_npi="2222222222",
+        service_date=date(2025, 1, 1),
+    )
+    matched = _content_keys_match(remit, claim)
+    assert "npi" in matched
+    assert "pcn" in matched
+    assert "charge" in matched
+
+
+def test_content_keys_match_npi_mismatch_does_not_match(db_session, make_claim, make_remit):
+    """SP32: NPI arm does not fire when NPIs differ."""
+    from cyclone.reconcile import _content_keys_match
+    from datetime import date
+    claim = make_claim(
+        patient_control_number="PCN2",
+        total_charge=Decimal("10.00"),
+        rendering_provider_npi="2222222222",
+        service_date_from=date(2025, 1, 1),
+    )
+    remit = make_remit(
+        payer_claim_control_number="PCN2",
+        total_charge_amount=Decimal("10.00"),
+        rendering_provider_npi="9999999999",  # different NPI
+        service_date=date(2025, 1, 1),
+    )
+    matched = _content_keys_match(remit, claim)
+    assert "npi" not in matched
+    assert "pcn" in matched
+    assert "charge" in matched

docs/superpowers/specs/2026-07-02-cyclone-rendering-npi-extraction-design.md

@@ -62,14 +62,22 @@ Effective algorithm unchanged: 2-of-3 `{PCN, charge, NPI}` still fires when any
 
 New subcommand: `python -m cyclone.cli backfill-rendering-npi`.
 
-Behavior:
+**Why explicit file args.** The `Batch` ORM does not retain a column for the original on-disk ingest path — the live ingest pipeline reads the upload body, parses, writes rows, and never persists the bytes. So there is no `claim_batches.inbound_path` / `remittances.inbound_path` to "look up" and re-parse from. The operator must point the CLI at the files (or the directory that holds them) that need backfill. Re-running ingest end-to-end would be the alternative, but it would re-`INSERT` claim rows and is a destructive operation; backfill is intentionally a read-reparse-targeted-`UPDATE`.
-1. For every `claim_batches.inbound_path`, re-parse the 837p file with the new parser; for every claim in the batch, write `Claim.rendering_provider_npi = parsed_value` **only when the column is currently `NULL`** (idempotent).
-2. For every `remittances.inbound_path`, re-parse the 835 file with the new parser; for every remit, write both the per-segment NPIs (`ClaimPayment.service_provider_npi`) and the aggregated `Remittance.rendering_provider_npi`.
-3. Run `reconcile.run()` once over all open pairs at the end so the NPI arm can fire retroactively.
-4. Log per-batch counts (parsed, updated, skipped).
-5. Exit code 0 on success, 2 on file-level failure (no batch updated), 1 on unexpected exception.
 
-Idempotent: re-running on a fully-populated DB is a no-op (typed columns not overwritten; reconcile already matched pairs skipped via `Claim.matched_remittance_id IS NOT NULL`).
+Flags:
+- `--file PATH` (repeatable) — one or more specific X12 files to re-parse. Each path is validated to exist before parsing.
+- `--input-dir DIR` — directory to scan one level deep for `*.txt` / `*.edi` / `*.x12` files. Falls back to `$CYCLONE_BACKFILL_INPUT_DIR` if neither `--file` nor `--input-dir` is passed.
+- `--type {837p,835}` (optional) — pin the parser. **When omitted, each file's transaction kind is auto-sniffed** (filename hint + ISA/ST prefix) so a mixed directory can be processed in one pass.
+
+Behavior:
+1. For each file resolved by `--file` / `--input-dir`, re-parse with the new parser (T4 wiring).
+2. **837p**: for every claim in the batch, write `Claim.rendering_provider_npi = parsed_value` **only when the column is currently `NULL`** (idempotent — already-populated rows are left alone).
+3. **835**: for every remit, write `Remittance.rendering_provider_npi` (typed column) **only when `NULL`**, and persist the per-segment `service_provider_npi` values into `Remittance.raw_json["service_provider_npis"]` (D5).
+4. After all files are processed, run `reconcile.run()` **once across every 835 Batch** so the D6 NPI arm can fire retroactively on newly-populated pairs.
+5. Log per-file counts (parsed, updated, skipped). Emit one-line summary to stdout: `claims_updated=N remits_updated=M files_processed=K files_skipped=J`.
+6. Exit code 0 on success (zero populated rows is still exit 0), 2 on file-level failure (no file processed), 1 on unexpected exception.
+
+Idempotent: re-running on a fully-populated DB is a no-op (typed columns not overwritten; reconcile skips already-matched pairs via `Claim.matched_remittance_id IS NOT NULL`).
 
 ### D8 — Backfill event emission
 
@@ -96,10 +104,11 @@ No `UPDATE` statements in the migration — backfill is a separate, deliberate s
 - `backend/src/cyclone/db.py` — `Claim.rendering_provider_npi`, `Remittance.rendering_provider_npi`. No claim_payments ORM — service_provider_npi persists only via `Remittance.raw_json["service_provider_npis"]` (D5).
 - `backend/src/cyclone/writer.py` / `writer_835.py` — set the new typed columns and the `raw_json` mirrors at write time. `writer_835.py` writes the per-CLP `service_provider_npi` into `Remittance.rendering_provider_npi` (D4, single value) and into `raw_json["service_provider_npis"]` (D5).
 - `backend/src/cyclone/reconcile.py` — extend `_content_keys_match` (D6).
-- `backend/src/cyclone/cli.py` — new `backfill-rendering-npi` subcommand.
+- `backend/src/cyclone/store/backfill.py` — **new helper module** owning `backfill_rendering_provider_npi(files, input_dir, transaction_type)`: file/CLI argument resolution, auto-sniffing (ISA/ST + filename hint), re-parse dispatch, idempotent typed-column writes, and the post-backfill `reconcile.run()` sweep. Returns a small summary dataclass for the CLI summary line.
+- `backend/src/cyclone/cli.py` — new `backfill-rendering-npi` subcommand (Click). Wires `--file` / `--input-dir` / `--type` / `--log-level`, delegates the heavy lifting to `store.backfill.backfill_rendering_provider_npi`, and prints the one-line summary.
 - `backend/src/cyclone/migrations/0019_add_rendering_and_service_provider_npis.sql` — D9.
 - `backend/tests/fixtures/` — new 837p fixture(s) with `NM1*82` and new 835 fixture(s) with `NM1*1P`.
-- `backend/tests/test_reconcile.py` — new tests for D6 (set-membership, comma-split, fallback chain).
+- `backend/tests/test_reconcile.py` — new tests for D6 (typed-column primary path, typed-column beats raw_json fallback, NPI arm fires when both sides populated).
 - `backend/tests/test_parse_835.py` — new test for `_consume_claim_payment` `NM1*1P` extraction.
 - `backend/tests/test_parse_837.py` — new test for rendering-provider loop extraction.
 
@@ -128,7 +137,7 @@ No `UPDATE` statements in the migration — backfill is a separate, deliberate s
 
 ### Unit tests (matcher level)
 
-- `test_reconcile.py`: comma-list set membership — single-CLP-remit with `Remittance.rendering_provider_npi = "1234567890,0987654321"` and `Claim.rendering_provider_npi = "0987654321"` counts NPI as a matched key.
+- `test_reconcile.py`: typed-column primary path — `Remittance.rendering_provider_npi = "1234567890"` and `Claim.rendering_provider_npi = "1234567890"` counts NPI as a matched key.
 - `test_reconcile.py`: multi-key fallback — `NM1*1P` NPI absent (`None`) on one side, present on the other → NPI arm does not fire, PCN+charge still matches (1-of-2 effective, but per SP31 the rule is 2-of-3 of {PCN, charge, NPI}, so PCN+charge alone does NOT auto-link — must stay 2 of 3 with NPI also matching or charge differing).
 - `test_reconcile.py`: typed-column primary path beats `raw_json` fallback — fixture sets both, matcher uses typed column.
 - `test_reconcile.py`: existing 22+ SP31 tests still pass without modification.