feat(pfs): multi-rule NPRM ingest — CY2026+CY2027 partitions (refs #612)

dev/scripts/ingest_pfs.py

@@ -36,14 +36,16 @@ def main() -> int:
     parser.add_argument(
         "--nprm",
         action="store_true",
-        help="also load the NPRM Addendum B proposed RVUs into pfs.rvu_proposed",
+        help=(
+            "also load every registered NPRM Addendum B proposed-RVU "
+            "partition (pfs.nprm.NPRM_SOURCES) into pfs.rvu_proposed"
+        ),
     )
     args = parser.parse_args()
 
     from cms import ingest_log
     from conf.connect import duckdb_batch, publish_replica
     from pfs.pipe import load_all
-    from pfs.rules import RULES
 
     run_id = ingest_log.new_run_id()
 
@@ -62,22 +64,23 @@ def main() -> int:
             )
 
         if args.nprm:
-            from pfs.nprm import load_rvu_proposed
+            from pfs.nprm import NPRM_SOURCES, load_rvu_proposed
 
             print("\n--- NPRM Addendum B (proposed RVUs) ---")
-            nprm_out = load_rvu_proposed(con)
+            for year, _tag, cms_rule_id, _pincite_key in NPRM_SOURCES:
-            print(f"  pfs.rvu_proposed        : {nprm_out}")
+                nprm_out = load_rvu_proposed(con, year=year)
-            ingest_log.log_ingest(
+                print(f"  pfs.rvu_proposed [{cms_rule_id}]: {nprm_out}")
-                con,
+                ingest_log.log_ingest(
-                module="pfs",
+                    con,
-                table_name="pfs.rvu_proposed",
+                    module="pfs",
-                rows=nprm_out["rows"],
+                    table_name="pfs.rvu_proposed",
-                source_file=nprm_out["source_file"],
+                    rows=nprm_out["rows"],
-                rule_id="CMS-1832-P",
+                    source_file=nprm_out["source_file"],
-                fr_citation=RULES[2026].proposed.federal_register_citation,
+                    rule_id=nprm_out["cms_rule_id"],
-                pincite_key="2KVJ2HKX",
+                    fr_citation=nprm_out["fr_citation"],
-                run_id=run_id,
+                    pincite_key=nprm_out["pincite_key"],
-            )
+                    run_id=run_id,
+                )
 
     replica = publish_replica("aco")
     print(f"replica → {replica}")

src/pfs/nprm.py

@@ -1,13 +1,18 @@
-"""NPRM Addendum B loader → ``pfs.rvu_proposed`` (#592).
+"""NPRM Addendum B loader → ``pfs.rvu_proposed`` (#592, #612).
 
-Loads the CY2026 NPRM Addendum B "Relative Value Units and Related
+Loads a PFS NPRM's Addendum B "Relative Value Units and Related
-Information" workbook — registered in bib under the
+Information" workbook — registered in bib under a per-year
-``sup:2026_PFS_NPRM`` tag — into ``pfs.rvu_proposed``, the
+``sup:<year>_PFS_NPRM`` tag — into ``pfs.rvu_proposed``, the
 proposed-rule counterpart to the final-rule ``pfs.rvu`` table that
 ``pfs.pipe.load_all`` populates. Reuses ``pfs.pipe``'s header-detect
 readers and column map instead of reimplementing them; the target
 schema is intentionally narrower than ``pfs.rvu`` (no PPRRVU-only
 columns) since Addendum B never carries them.
+
+``pfs.rvu_proposed`` holds one partition per ``cms_rule_id`` — CY2026's
+NPRM (CMS-1832-P) and CY2027's NPRM (CMS-1848-P) load side by side, and
+``load_rvu_proposed`` only ever deletes/reinserts the partition for the
+year it was called with (see ``NPRM_SOURCES`` below).
 """
 
 from __future__ import annotations
@@ -16,12 +21,16 @@ import re
 from pathlib import Path
 from typing import Any
 
-_ADDENDUM_B_TAG = "sup:2026_PFS_NPRM"
 _ADDENDUM_B_RE = re.compile(r"(?i)addendum[_ ]?b")
 
 # pfs.pipe._RVU_COLUMNS normalizes raw headers to these intermediate
 # names, which already match the pfs.rvu_proposed target schema (#607
 # aligned it with pfs.rvu's column names — no renaming needed here).
+# The CY2027 Addendum B workbook's headers (CPT1/\nHCPCS, Mod, Status,
+# \nWork\nRVUs2, \nNon-\nFacility\nPE\nRVUs2, \nFacility\nPE\nRVUs2,
+# Mal-\nPractice\nRVUs2) are already present in _RVU_COLUMNS as
+# variants of the CY2026 PPRRVU headers, so no map changes were needed
+# for #612 — verified against the real CY2027 file.
 _TARGET_COLUMNS = [
     "hcpcs",
     "mod",
@@ -37,13 +46,31 @@ _TARGET_COLUMNS = [
 # Intermediate (pre-rename) names as produced by pfs.pipe._normalize_columns.
 _RAW_NUMERIC_COLUMNS = ("work_rvu", "non_fac_pe_rvu", "fac_pe_rvu", "mp_rvu")
 
+# (year, sup_tag, cms_rule_id, pincite_key) — one entry per NPRM Addendum
+# B partition. Add a row here (plus a pincite/proposed_for entry, see
+# pfs.rules) to onboard a new year; load_rvu_proposed(year=...) and
+# dev/scripts/ingest_pfs.py --nprm both iterate this tuple.
+NPRM_SOURCES: tuple[tuple[int, str, str, str], ...] = (
+    (2026, "sup:2026_PFS_NPRM", "CMS-1832-P", "2KVJ2HKX"),
+    (2027, "sup:2027_PFS_NPRM", "CMS-1848-P", "5ITGVDJV"),
+)
 
-def _discover_addendum_b() -> Path:
+
+def _source_for(year: int) -> tuple[int, str, str, str]:
+    """Look up ``year``'s ``NPRM_SOURCES`` entry, or raise ``ValueError``."""
+    for source in NPRM_SOURCES:
+        if source[0] == year:
+            return source
+    known = ", ".join(str(s[0]) for s in NPRM_SOURCES)
+    raise ValueError(f"No NPRM source registered for year {year} (known: {known})")
+
+
+def _discover_addendum_b(tag: str) -> Path:
     """Locate the NPRM Addendum B attachment in the bib store.
 
-    Queries attachments of items tagged ``sup:2026_PFS_NPRM`` for one
+    Queries attachments of items tagged ``tag`` for one whose filename
-    whose filename matches ``addendum[_ ]?b`` (case-insensitive),
+    matches ``addendum[_ ]?b`` (case-insensitive), mirroring the join
-    mirroring the join in ``llm/source.py::_attachment_text``.
+    in ``llm/source.py::_attachment_text``.
     """
     from conf import connect
 
@@ -57,7 +84,7 @@ def _discover_addendum_b() -> Path:
             "JOIN tags t ON t.id = it.tag_id "
             "WHERE t.name = ? "
             "ORDER BY a.id",
-            (_ADDENDUM_B_TAG,),
+            (tag,),
         )
         .fetchall()
     )
@@ -66,17 +93,43 @@ def _discover_addendum_b() -> Path:
         if _ADDENDUM_B_RE.search(filename):
             return Path(row["storage_path"])
     raise FileNotFoundError(
-        f"No Addendum B attachment found among bib items tagged {_ADDENDUM_B_TAG!r}"
+        f"No Addendum B attachment found among bib items tagged {tag!r}"
     )
 
 
-def load_rvu_proposed(con: Any, *, cms_rule_id: str = "CMS-1832-P") -> dict:
+def _schema_matches_target(con: Any) -> bool:
-    """Load the NPRM Addendum B workbook into ``pfs.rvu_proposed``.
+    """True if ``pfs.rvu_proposed`` doesn't exist yet, or its columns match.
 
-    Idempotent: deletes any existing rows for ``cms_rule_id`` before
+    False means a stale table (e.g. the pre-#612 ``modifier`` /
-    inserting, so re-running the ingest is safe.
+    ``nonfac_pe_rvu`` column names) needs dropping before the
+    ``CREATE TABLE IF NOT EXISTS`` below — that statement is a no-op
+    against an existing table, so it can't fix a column mismatch on
+    its own.
+    """
+    rows = con.execute(
+        "SELECT column_name FROM information_schema.columns "
+        "WHERE table_schema = 'pfs' AND table_name = 'rvu_proposed' "
+        "ORDER BY ordinal_position"
+    ).fetchall()
+    if not rows:
+        return True
+    existing = [r[0] for r in rows]
+    return existing == _TARGET_COLUMNS
 
-    Returns ``{"rows": int, "source_file": str}``.
+
+def load_rvu_proposed(con: Any, *, year: int) -> dict:
+    """Load ``year``'s NPRM Addendum B workbook into ``pfs.rvu_proposed``.
+
+    Resolves the bib tag, ``cms_rule_id``, and pincite key from
+    ``NPRM_SOURCES``, and the Federal Register citation from
+    ``pfs.rules.proposed_for(year)``.
+
+    Idempotent: deletes any existing rows for that year's
+    ``cms_rule_id`` before inserting, so re-running the ingest is safe
+    and leaves other years' partitions untouched.
+
+    Returns ``{"rows": int, "source_file": str, "cms_rule_id": str,
+    "fr_citation": str, "pincite_key": str}``.
     """
     import polars as pl
 
@@ -86,8 +139,13 @@ def load_rvu_proposed(con: Any, *, cms_rule_id: str = "CMS-1832-P") -> dict:
         _read_csv_with_header,
         _read_excel,
     )
+    from pfs.rules import proposed_for
 
-    path = _discover_addendum_b()
+    _, tag, cms_rule_id, pincite_key = _source_for(year)
+    rule = proposed_for(year)
+    fr_citation = rule.federal_register_citation if rule is not None else ""
+
+    path = _discover_addendum_b(tag)
     ext = path.suffix.lower()
     if ext == ".csv":
         df = _read_csv_with_header(str(path))
@@ -133,6 +191,8 @@ def load_rvu_proposed(con: Any, *, cms_rule_id: str = "CMS-1832-P") -> dict:
     df = df.select(_TARGET_COLUMNS)
 
     con.execute("CREATE SCHEMA IF NOT EXISTS pfs")
+    if not _schema_matches_target(con):
+        con.execute("DROP TABLE IF EXISTS pfs.rvu_proposed")
     con.execute(
         """
         CREATE TABLE IF NOT EXISTS pfs.rvu_proposed (
@@ -159,4 +219,10 @@ def load_rvu_proposed(con: Any, *, cms_rule_id: str = "CMS-1832-P") -> dict:
     finally:
         con.unregister("_nprm_df")
 
-    return {"rows": df.height, "source_file": str(path)}
+    return {
+        "rows": df.height,
+        "source_file": str(path),
+        "cms_rule_id": cms_rule_id,
+        "fr_citation": fr_citation,
+        "pincite_key": pincite_key,
+    }

tests/pfs/test_nprm.py

@@ -1,14 +1,15 @@
-"""Tests for pfs.nprm — NPRM Addendum B loader → pfs.rvu_proposed (#592)."""
+"""Tests for pfs.nprm — NPRM Addendum B loader → pfs.rvu_proposed (#592, #612)."""
 
 from __future__ import annotations
 
 import duckdb
+import pytest
 
 from pfs import nprm
 
 
-def _fixture(tmp_path):
+def _fixture(tmp_path, name="CY2026_NPRM_Addendum_B.csv"):
-    p = tmp_path / "CY2026_NPRM_Addendum_B.csv"
+    p = tmp_path / name
     p.write_text(
         "HCPCS,MOD,DESCRIPTION,STATUS CODE,WORK RVU,"
         "NON-FAC PE RVU,FAC PE RVU,MP RVU\n"
@@ -18,20 +19,135 @@ def _fixture(tmp_path):
     return p
 
 
+def _fixture_2027(tmp_path):
+    """CY2027 partition with a distinct HCPCS set so the two are distinguishable."""
+    p = tmp_path / "CY2027_NPRM_Addendum_B.csv"
+    p.write_text(
+        "HCPCS,MOD,DESCRIPTION,STATUS CODE,WORK RVU,"
+        "NON-FAC PE RVU,FAC PE RVU,MP RVU\n"
+        "99214,,Office visit est L4,A,1.9,2.1,0.75,0.15\n"
+        "0002A,,Admin covid dose 2,X,0.0,0.0,0.0,0.0\n"
+    )
+    return p
+
+
+def _patch_discovery(monkeypatch, mapping):
+    """monkeypatch._discover_addendum_b(tag) -> path, from a {tag: path} dict."""
+
+    def _fake(tag):
+        try:
+            return mapping[tag]
+        except KeyError:
+            raise FileNotFoundError(
+                f"No Addendum B attachment found among bib items tagged {tag!r}"
+            ) from None
+
+    monkeypatch.setattr(nprm, "_discover_addendum_b", _fake)
+
+
 def test_load_rvu_proposed_loads_and_reloads(tmp_path, monkeypatch):
     path = _fixture(tmp_path)
-    monkeypatch.setattr(nprm, "_discover_addendum_b", lambda: path)
+    _patch_discovery(monkeypatch, {"sup:2026_PFS_NPRM": path})
     con = duckdb.connect()
-    out = nprm.load_rvu_proposed(con)
+    out = nprm.load_rvu_proposed(con, year=2026)
     assert out["rows"] == 2
+    assert out["cms_rule_id"] == "CMS-1832-P"
     row = con.execute(
         "SELECT hcpcs, work_rvu, cms_rule_id FROM pfs.rvu_proposed WHERE hcpcs='99213'"
     ).fetchone()
     assert row == ("99213", 1.3, "CMS-1832-P")
-    nprm.load_rvu_proposed(con)  # idempotent delete-and-reload
+    nprm.load_rvu_proposed(con, year=2026)  # idempotent delete-and-reload
     assert con.execute("SELECT count(*) FROM pfs.rvu_proposed").fetchone()[0] == 2
 
 
+def test_nprm_sources_covers_both_years():
+    years = {source[0] for source in nprm.NPRM_SOURCES}
+    assert years == {2026, 2027}
+
+
+def test_load_rvu_proposed_two_partitions_side_by_side(tmp_path, monkeypatch):
+    """Loading both years leaves both rule partitions intact side by side."""
+    path_2026 = _fixture(tmp_path)
+    path_2027 = _fixture_2027(tmp_path)
+    _patch_discovery(
+        monkeypatch,
+        {"sup:2026_PFS_NPRM": path_2026, "sup:2027_PFS_NPRM": path_2027},
+    )
+    con = duckdb.connect()
+    nprm.load_rvu_proposed(con, year=2026)
+    nprm.load_rvu_proposed(con, year=2027)
+
+    counts = dict(
+        con.execute(
+            "SELECT cms_rule_id, count(*) FROM pfs.rvu_proposed GROUP BY 1"
+        ).fetchall()
+    )
+    assert counts == {"CMS-1832-P": 2, "CMS-1848-P": 2}
+
+
+def test_load_rvu_proposed_reload_one_leaves_other_intact(tmp_path, monkeypatch):
+    """Reloading the 2026 partition doesn't touch the 2027 partition's rows."""
+    path_2026 = _fixture(tmp_path)
+    path_2027 = _fixture_2027(tmp_path)
+    _patch_discovery(
+        monkeypatch,
+        {"sup:2026_PFS_NPRM": path_2026, "sup:2027_PFS_NPRM": path_2027},
+    )
+    con = duckdb.connect()
+    nprm.load_rvu_proposed(con, year=2026)
+    nprm.load_rvu_proposed(con, year=2027)
+    nprm.load_rvu_proposed(con, year=2026)  # reload just 2026
+
+    counts = dict(
+        con.execute(
+            "SELECT cms_rule_id, count(*) FROM pfs.rvu_proposed GROUP BY 1"
+        ).fetchall()
+    )
+    assert counts == {"CMS-1832-P": 2, "CMS-1848-P": 2}
+    row_2027 = con.execute(
+        "SELECT hcpcs, work_rvu FROM pfs.rvu_proposed WHERE cms_rule_id='CMS-1848-P' "
+        "AND hcpcs='99214'"
+    ).fetchone()
+    assert row_2027 == ("99214", 1.9)
+
+
+def test_load_rvu_proposed_missing_tag_raises_naming_it(tmp_path, monkeypatch):
+    """Discovery failure names the missing tag in the error."""
+    _patch_discovery(monkeypatch, {})  # no tags registered at all
+    con = duckdb.connect()
+    with pytest.raises(FileNotFoundError, match="sup:2026_PFS_NPRM"):
+        nprm.load_rvu_proposed(con, year=2026)
+
+
+def test_load_rvu_proposed_unknown_year_raises():
+    con = duckdb.connect()
+    with pytest.raises(ValueError, match="2099"):
+        nprm.load_rvu_proposed(con, year=2099)
+
+
+def test_load_rvu_proposed_drops_and_recreates_stale_schema(tmp_path, monkeypatch):
+    """A pre-existing table with old column names is dropped and rebuilt.
+
+    Regression guard for the pre-#612 table (columns ``modifier`` /
+    ``nonfac_pe_rvu`` instead of ``mod`` / ``non_fac_pe_rvu``) — CREATE
+    TABLE IF NOT EXISTS alone won't fix a live mismatched table.
+    """
+    path = _fixture(tmp_path)
+    _patch_discovery(monkeypatch, {"sup:2026_PFS_NPRM": path})
+    con = duckdb.connect()
+    con.execute("CREATE SCHEMA IF NOT EXISTS pfs")
+    con.execute(
+        "CREATE TABLE pfs.rvu_proposed ("
+        "hcpcs VARCHAR, modifier VARCHAR, description VARCHAR, "
+        "status_code VARCHAR, work_rvu DOUBLE, nonfac_pe_rvu DOUBLE, "
+        "fac_pe_rvu DOUBLE, mp_rvu DOUBLE, cms_rule_id VARCHAR)"
+    )
+    out = nprm.load_rvu_proposed(con, year=2026)
+    assert out["rows"] == 2
+    cols = [r[0] for r in con.execute("DESCRIBE pfs.rvu_proposed").fetchall()]
+    assert cols == nprm._TARGET_COLUMNS
+
+
 def _dirty_fixture(tmp_path):
     """Real-file quirks: padded cells, "NA" markers, a footnote row.
 
@@ -55,9 +171,9 @@ def _dirty_fixture(tmp_path):
 def test_load_rvu_proposed_strips_padding(tmp_path, monkeypatch):
     """Whitespace-padded HCPCS/numeric cells load clean, not null."""
     path = _dirty_fixture(tmp_path)
-    monkeypatch.setattr(nprm, "_discover_addendum_b", lambda: path)
+    _patch_discovery(monkeypatch, {"sup:2026_PFS_NPRM": path})
     con = duckdb.connect()
-    nprm.load_rvu_proposed(con)
+    nprm.load_rvu_proposed(con, year=2026)
     row = con.execute(
         "SELECT hcpcs, description, work_rvu, non_fac_pe_rvu, fac_pe_rvu, mp_rvu "
         "FROM pfs.rvu_proposed WHERE hcpcs='99213'"
@@ -68,9 +184,9 @@ def test_load_rvu_proposed_strips_padding(tmp_path, monkeypatch):
 def test_load_rvu_proposed_na_marker_becomes_null(tmp_path, monkeypatch):
     """ "NA" in a numeric cell loads as SQL NULL, not a cast error."""
     path = _dirty_fixture(tmp_path)
-    monkeypatch.setattr(nprm, "_discover_addendum_b", lambda: path)
+    _patch_discovery(monkeypatch, {"sup:2026_PFS_NPRM": path})
     con = duckdb.connect()
-    nprm.load_rvu_proposed(con)
+    nprm.load_rvu_proposed(con, year=2026)
     row = con.execute(
         "SELECT work_rvu, non_fac_pe_rvu, fac_pe_rvu, mp_rvu "
         "FROM pfs.rvu_proposed WHERE hcpcs='99214'"
@@ -81,9 +197,9 @@ def test_load_rvu_proposed_na_marker_becomes_null(tmp_path, monkeypatch):
 def test_load_rvu_proposed_drops_footnote_rows(tmp_path, monkeypatch):
     """A footnote row (first cell = long sentence) is not a HCPCS code."""
     path = _dirty_fixture(tmp_path)
-    monkeypatch.setattr(nprm, "_discover_addendum_b", lambda: path)
+    _patch_discovery(monkeypatch, {"sup:2026_PFS_NPRM": path})
     con = duckdb.connect()
-    out = nprm.load_rvu_proposed(con)
+    out = nprm.load_rvu_proposed(con, year=2026)
     assert out["rows"] == 2  # only 99213 and 99214 — footnote row excluded
     assert con.execute("SELECT count(*) FROM pfs.rvu_proposed").fetchone()[0] == 2
     footnote = con.execute(