From 85663894f393c8b8adf7b25d67916abe8a78205c Mon Sep 17 00:00:00 2001 From: kert Date: Thu, 12 Nov 2026 04:00:47 -0500 Subject: [PATCH] =?UTF-8?q?fix(pfs):=20origin=20=E2=80=94=20deletion=20row?= =?UTF-8?q?s=20from=20lineage;=20final-rule=20coverage=20rows=20tell=20a?= =?UTF-8?q?=20finalized=20removal;=20CFR=20part=20check=20for=20title=2042?= =?UTF-8?q?=20only?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A family code with deleted + cpt_panel events at one FR paragraph gets a cpt_panel row 'The CPT Editorial Panel deleted , effective CY.' (extractor lineage) where a plain sentence reads as the Panel deleting it and the paragraph writes the year; a model Panel row from the same rule that tells a lineage-deleted code as merged, revised or retained is dropped. IOM/MLN rows cited from FR paragraphs count only from final rules; a provision whose removal the rule finalizes becomes a row saying what was removed and when, citing the finalizing paragraph (ref_p_id = the citing one). Only title-42 parts outside the cited chapters are FR typos. --- src/pfs/origin.py | 295 ++++++++++++++++++++++++++++++--- tests/pfs/test_origin.py | 350 ++++++++++++++++++++++++++++++++++++++- 2 files changed, 619 insertions(+), 26 deletions(-) diff --git a/src/pfs/origin.py b/src/pfs/origin.py index cf7f771..13f3798 100644 --- a/src/pfs/origin.py +++ b/src/pfs/origin.py @@ -15,7 +15,14 @@ Three sources, every row citing the exact place that supports it: paraphrase (no 8-word run shared with the page), is at most 240 characters, and the ledger stores only the guarded paraphrases. * ``pfs.code_guidance`` rows (deterministic): CFR cites become - ``regulation`` rows, IOM and MLN cites ``coverage`` rows. + ``regulation`` rows, IOM and MLN cites ``coverage`` rows (an FR-cited + manual provision only from a final rule; one whose removal that rule + finalizes says so and cites the finalizing paragraph). +* ``pfs.code_event`` (deterministic, ``extractor="lineage"``): a family + code the CPT Editorial Panel deleted (``deleted`` + ``cpt_panel`` at one + paragraph) gets a ``cpt_panel`` row stating its deletion, and a model + row from the same rule that tells it as merged, revised or retained is + dropped. """ from __future__ import annotations @@ -25,10 +32,11 @@ from collections.abc import Collection, Iterator, Sequence from pathlib import Path from typing import Any -from pfs.codetables import ORIGIN_KINDS, OriginRow, read_guidance +from bib.frlink import rule_kind as frlink_rule_kind +from pfs.codetables import ORIGIN_KINDS, OriginRow, read_events, read_guidance from pfs.descriptors import rule_year_of from pfs.families import code_pattern -from pfs.lineage import _expand_ranges +from pfs.lineage import _expand_ranges, deletion_readings from pfs.rounds import ( _HYPOTHETICAL, _PRINTED_PAGE, @@ -671,35 +679,158 @@ def _removed(store: Any, item_key: str, p_id: int, locator: str) -> bool: ) -#: 42 CFR chapter IV (CMS) is parts 400-699; any other part is a typo in the -#: FR text (a "42 CFR 25.400" for 425.400) -_CMS_PARTS = range(400, 700) +#: the 42 CFR parts the rules cite: chapter IV (CMS, 400-699), chapter V +#: (OIG, 1000-1008) and the chapter I parts 2 (substance use disorder +#: records), 8 (opioid treatment programs), 10 (340B), 51 and 56. Any other +#: title-42 part is a typo in the FR text ("42 CFR 25.400" for 425.400); +#: other titles (45 CFR 170, 21 CFR 882, ...) are not checked. +_TITLE42_PARTS = frozenset({2, 8, 10, 51, 56, *range(400, 700), *range(1000, 1009)}) + + +def _real_cfr(locator: str) -> bool: + m = re.match(r"(\d+) CFR (\d+)", locator) + return not (m and m.group(1) == "42" and int(m.group(2)) not in _TITLE42_PARTS) + + +#: "remove the requirement that the medical record must document ..." +_REQUIREMENT = re.compile( + r"\b(?:remov|eliminat|rescind)\w*\s+(?:the|this)\s+requirement\s+that\s+" + r"(?P[^.;,]+)", + re.I, +) +_EFFECTIVE = re.compile(r"\beffective\s+([A-Z][a-z]+ \d{1,2}, \d{4})") +#: how many paragraphs after the citing one a finalizing response may sit +_RESPONSE_SPAN = 5 + + +def _rule_meta(store: Any, item_key: str, cache: dict) -> tuple[str, int | None, str]: + """``(rule kind, rule year, FR volume)`` of a bib item, cached.""" + if item_key not in cache: + con = store._con() # noqa: SLF001 + row = con.execute( + "SELECT title, date_published FROM items WHERE key = ?", (item_key,) + ).fetchone() + title, dp = row if row else ("", "") + vol = con.execute( + "SELECT fr_volume FROM fr_anchor_docs WHERE item_key = ?", (item_key,) + ).fetchone() + cache[item_key] = ( + frlink_rule_kind(store, item_key, title=title or ""), + rule_year_of(title or "", dp or ""), + str(vol[0]) if vol and vol[0] else "", + ) + return cache[item_key] + + +def _fr_cite(volume: str, page: int, p_id: int, ref_p_id: int) -> str: + cite = f"{volume} FR {page}" if volume else f"FR p. {page}" + note = f" (reference ¶{ref_p_id})" if ref_p_id != p_id else "" + return f"{cite}, ¶{p_id}{note}" + + +def _finalized_removal( + store: Any, item_key: str, p_id: int, rule_year: int | None +) -> dict | None: + """The finalized removal of a requirement the citing paragraph *p_id* + describes ("... remove the requirement that X"): a plain sentence that + finalizes removing the same requirement, in that paragraph or one of + the next five of the rule. ``{"p_id", "page", "req", "effective"}``.""" + rows = ( + store._con() # noqa: SLF001 + .execute( + "SELECT p_id, page, text FROM fr_anchors WHERE item_key = ? " + "AND p_id BETWEEN ? AND ? ORDER BY p_id", + (item_key, p_id, p_id + _RESPONSE_SPAN), + ) + .fetchall() + ) + if not rows or rows[0][0] != p_id: + return None + reqs = {_collapse(m.group("req")) for m in _REQUIREMENT.finditer(rows[0][2] or "")} + if not reqs: + return None + for q, page, text in rows: + for sent in _sentences(text or ""): + m = _REQUIREMENT.search(sent) + if ( + m + and _collapse(m.group("req")) in reqs + and _FINALIZING.search(sent) + and _plain(sent, rule_year) + ): + eff = _EFFECTIVE.search(sent) + return { + "p_id": q, + "page": page, + "req": " ".join(m.group("req").split()), + "effective": eff.group(1) if eff else "", + } + return None + + +def _removal_row(g: Any, locator: str, meta: tuple, found: dict) -> OriginRow: + """The coverage row of a manual provision whose removal a final rule + finalized: what was removed and when, citing the finalizing paragraph + (``ref_p_id`` = the paragraph that cites the provision).""" + _kind, year, volume = meta + cy = f"CY{year} " if year else "" + when = f", effective {found['effective']}" if found["effective"] else "" + summary = ( + f"The {cy}final rule removed the requirement in {locator} that " + f"{found['req']}{when}." + ) + if len(summary) > 240: + summary = f"The {cy}final rule removed a requirement in {locator}{when}." + return OriginRow( + family=g.family, + code=g.code, + kind="coverage", + reference=locator, + year=year, + summary=summary, + cite_kind=g.kind, + item_key=g.item_key_src, + p_id=found["p_id"], + page_label=str(found["page"]), + locator=f"{locator}; " + + _fr_cite(volume, found["page"], found["p_id"], g.p_id_src), + extractor="guidance", + ref_p_id=g.p_id_src, + ) def guidance_origins(con: Any, family: str, store: Any = None) -> list[OriginRow]: """One ``regulation`` (CFR section, sub-paragraph variants collapsed) or ``coverage`` (IOM, MLN) row per distinct ``(code, cite)`` in ``pfs.code_guidance``, citing the first paragraph that carries it. Not - origins: telehealth-listing CFR sections, and (with a *store* to read - the citing paragraph) manual provisions that paragraph is removing.""" + origins: telehealth-listing CFR sections and title-42 parts that do + not exist. With a *store* (to read the citing paragraph): a manual or + MLN cite from an FR paragraph counts only from a final rule; a + provision whose removal that rule finalizes becomes a row saying so + (citing the finalizing paragraph); one the citing sentence removes + without a finalized removal is dropped.""" rows: dict[tuple[str, str, str], OriginRow] = {} + rules: dict[str, tuple[str, int | None, str]] = {} for g in read_guidance(con, family): locator = g.locator if g.kind == "cfr": m = _CFR_SECTION.match(locator) locator = m.group(1) if m else locator - part = re.match(r"\d+ CFR (\d+)", locator) - if part and int(part.group(1)) not in _CMS_PARTS: + if not _real_cfr(locator): continue key = (g.code, g.kind, locator) if key in rows or locator in _NOT_ORIGINS: continue - if ( - store is not None - and g.kind == "iom" - and _removed(store, g.item_key_src, g.p_id_src, locator) - ): - continue + if store is not None and g.kind != "cfr" and g.p_id_src: + meta = _rule_meta(store, g.item_key_src, rules) + if meta[0] != "final": + continue + found = _finalized_removal(store, g.item_key_src, g.p_id_src, meta[1]) + if found is not None: + rows[key] = _removal_row(g, locator, meta, found) + continue + if g.kind == "iom" and _removed(store, g.item_key_src, g.p_id_src, locator): + continue rows[key] = OriginRow( family=family, code=g.code, @@ -831,9 +962,7 @@ def _locator(job: dict, it: dict) -> tuple[str, str]: else: page, text = job["page"], job["text"] page = _page_of(text, it["anchor"], page) - cite = f"{job['volume']} FR {page}" if job["volume"] else f"FR p. {page}" - note = f" (reference ¶{ref_p})" if ref_p else "" - return str(page), f"{cite}, ¶{job['p_id']}{note}" + return str(page), _fr_cite(job["volume"], page, job["p_id"], ref_p or job["p_id"]) def _rows( @@ -895,12 +1024,128 @@ def _consolidate(rows: list[tuple[OriginRow, dict]]) -> list[OriginRow]: else: ref = _ACT_SUFFIX.sub("", _AMA.sub("", it["anchor"])).casefold() key = (row.code, row.kind, ref, row.year) - rank = (job["rule_kind"] != "final", -len(it["codes"]), n) + rank = ( + job["rule_kind"] != "final", + job["source"] != "lineage", + not it.get("states_year", False), + -len(it["codes"]), + n, + ) if key not in best or rank < best[key][0]: best[key] = (rank, row) return [row for _rank, row in best.values()] +# ── deletions, from lineage ────────────────────────────────────────── + +_PANEL = "CPT Editorial Panel" +_DELETION = re.compile(r"\bdelet\w*", re.I) +#: a fate other than deletion: merged into, revised, retained +_KEPT = re.compile(r"\b(?:merg|revis|retain|remain|combin|renam|kept)\w*", re.I) + + +def _fr_events(con: Any, codes: Sequence[str]) -> dict[str, list[Any]]: + """Each code's Federal Register events in ``pfs.code_event``.""" + return {c: [e for e in read_events(con, c) if e.source == "fr"] for c in codes} + + +def misstates_deletion(summary: str, code: str) -> bool: + """True when *summary* does not tell that *code* was deleted: it names + the code without saying it was deleted (a merge target, a revised + code), or it names no code and tells the codes only as merged, revised + or retained.""" + if _names(summary, code): + return "deleted" not in deletion_readings(summary, code) + return not _DELETION.search(summary) and bool(_KEPT.search(summary)) + + +def _drop_misstated( + rows: list[tuple[OriginRow, dict]], events: dict[str, list[Any]] +) -> list[tuple[OriginRow, dict]]: + """Model Panel rows less a code the same rule's lineage has deleted + when the row tells that code's fate otherwise.""" + deleted = { + (c, e.item_key) for c, evs in events.items() for e in evs if e.kind == "deleted" + } + return [ + (row, meta) + for row, meta in rows + if not ( + row.kind == "cpt_panel" + and row.cite_kind == "fr_para" + and (row.code, row.item_key) in deleted + and misstates_deletion(meta["item"]["summary"], row.code) + ) + ] + + +def lineage_origins( + store: Any, family: Any, events: dict[str, list[Any]] +) -> list[tuple[OriginRow, dict]]: + """A ``cpt_panel`` row "The CPT Editorial Panel deleted , + effective CY." for each family code with ``deleted`` and + ``cpt_panel`` events at one FR paragraph (*events*: ``_fr_events``), + citing that paragraph. Only where the paragraph supports it: a plain + sentence (no recap, hypothetical, request or proposal) reads as the + Panel deleting the code, the paragraph writes the year, and the rule + is no correction. Rows sharing ``(code, kind, reference, year)`` are + consolidated later; a sentence that writes the year ranks first.""" + at: dict[tuple[str, int], dict[str, set[str]]] = {} + years: dict[tuple[str, int], int] = {} + for code, evs in events.items(): + for e in evs: + at.setdefault((e.item_key, e.p_id), {}).setdefault(e.kind, set()).add(code) + years[(e.item_key, e.p_id)] = e.year + rules: dict[str, tuple[str, int | None, str]] = {} + con = store._con() # noqa: SLF001 + out: list[tuple[OriginRow, dict]] = [] + for (item_key, p_id), kinds in sorted(at.items()): + codes = sorted(kinds.get("deleted", set()) & kinds.get("cpt_panel", set())) + if not codes: + continue + para = con.execute( + "SELECT page, text FROM fr_anchors WHERE item_key = ? AND p_id = ?", + (item_key, p_id), + ).fetchone() + if para is None: + continue + page, text = para[0], para[1] or "" + rule_kind, rule_year, volume = _rule_meta(store, item_key, rules) + year = years[(item_key, p_id)] + if rule_kind == "correction" or str(year) not in _NUM.findall(text): + continue + for code in codes: + sents = [ + s + for s in _sentences(text) + if deletion_readings(s, code) == {"deleted", "cpt_panel"} + and _plain(s, rule_year) + ] + if not sents: + continue + page_label = _page_of(text, " ".join(sents[0].split()[:6]), page) + row = OriginRow( + family=family.key, + code=code, + kind="cpt_panel", + reference=_PANEL, + year=year, + summary=f"The {_PANEL} deleted {code}, effective CY{year}.", + cite_kind="fr_para", + item_key=item_key, + p_id=p_id, + page_label=str(page_label), + locator=_fr_cite(volume, page_label, p_id, p_id), + extractor="lineage", + ref_p_id=p_id, + ) + states = any(str(year) in _NUM.findall(s) for s in sents) + job = {"source": "lineage", "rule_kind": rule_kind} + item = {"anchor": _PANEL, "codes": codes, "states_year": states} + out.append((row, {"job": job, "item": item})) + return out + + def family_origin( con: Any, store: Any, @@ -917,9 +1162,12 @@ def family_origin( ``books`` is ``(item_key, edition_label, epub path)``. The model runs through the ledger *log* (resumable, ``limit`` caps new calls); without *ask* nothing is asked but the finished ledger for *replay_model* is - replayed, so ``--write`` never drops model rows. Model rows are - consolidated by ``_consolidate``. Returns ``(rows, run counts)``.""" + replayed, so ``--write`` never drops model rows. Deletion rows come + from lineage (``lineage_origins``; a model Panel row that misstates a + deleted code is dropped), and model and lineage rows are consolidated + by ``_consolidate``. Returns ``(rows, run counts)``.""" counts = {"done": 0, "skipped": 0, "failed": 0, "failed_pending": 0} + events = _fr_events(con, family.codes) model_rows: list[tuple[OriginRow, dict]] = [] model = ask.model if ask is not None else replay_model if model is not None: @@ -941,7 +1189,8 @@ def family_origin( entry = log.get(j["unit"], j["hash"], model, ORIGIN_PROMPT_VERSION) if entry is not None and entry.output is not None: model_rows += _rows(j, entry.output, family, model) - rows = _consolidate(model_rows) + guidance_origins(con, family.key, store) + found = _drop_misstated(model_rows, events) + lineage_origins(store, family, events) + rows = _consolidate(found) + guidance_origins(con, family.key, store) rows.sort( key=lambda r: (r.code, r.kind, r.year or 0, r.reference, r.item_key, r.p_id) ) diff --git a/tests/pfs/test_origin.py b/tests/pfs/test_origin.py index 64cbdd5..2daed4d 100644 --- a/tests/pfs/test_origin.py +++ b/tests/pfs/test_origin.py @@ -466,7 +466,7 @@ def test_guidance_drops_manual_provisions_the_citing_sentence_removes(tmp_path): 50, 1, 50, - "We propose to remove the rule in Section 30.6.14.1 of the manual.", + "We are removing the rule in Section 30.6.14.1 of the manual.", ), ( key, @@ -502,8 +502,8 @@ def test_guidance_drops_manual_provisions_the_citing_sentence_removes(tmp_path): ], ) kept = {(r.code, r.reference) for r in guidance_origins(c, "HOME-VISITS", s)} - # removal must be in the sentence that cites the section; a CFR part - # outside 400-699 is an FR typo + # removal must be in the sentence that cites the section; a title-42 + # CFR part that does not exist is an FR typo assert kept == { ("99341", "100-04 ch.12 §30.6.14.2"), ("99341", "100-04 ch.26"), @@ -513,6 +513,350 @@ def test_guidance_drops_manual_provisions_the_citing_sentence_removes(tmp_path): } +def test_the_cfr_part_check_is_for_title_42_only(): + c = _con() + cites = [ + "45 CFR 170.315(b)", + "45 CFR 164.502", + "21 CFR 882.5801", + "42 CFR 8.12", + "42 CFR 2.11", + "42 CFR 10.10", + "42 CFR 1001.952(h)", + "42 CFR 1008.1", + "42 CFR 425.20", + "42 CFR 25.400(c)(1)(vi)", + "42 CFR 160.103", + ] + write_guidance(c, "HOME-VISITS", [_g("99341", "cfr", x) for x in cites]) + kept = {r.reference for r in guidance_origins(c, "HOME-VISITS")} + assert kept == { + "45 CFR 170.315", + "45 CFR 164.502", + "21 CFR 882.5801", + "42 CFR 8.12", + "42 CFR 2.11", + "42 CFR 10.10", + "42 CFR 1001.952", + "42 CFR 1008.1", + "42 CFR 425.20", + } + + +# public FR text: CY 2019 PFS proposed rule (83 FR 35835, CWGP3WJG ¶702) +# and final rule (83 FR 59630, 9V89NVZG ¶1622-¶1624), CY 2023 final +# (PETM225D ¶3607) +P702 = ( + "Medicare pays for E/M visits furnished in the home (a private residence) " + "under CPT codes 99341 through 99350. However, there is a Medicare Claims " + "Processing Manual provision requiring that the medical record must " + "document the medical necessity of the home visit made in lieu of an office " + "or outpatient visit (Pub. 100-04, Medicare Claims Processing Manual, " + "Chapter 12, Section 30.6.14.1.B). We agree, so we are proposing to remove " + "the requirement that the medical record must document the medical " + "necessity of furnishing the visit in the home rather than in the office." +) +P1622 = P702.replace("We agree, so we are proposing", "We agreed, so we proposed") +P1623 = ( + "Comment: Commenters were generally supportive of our proposal to remove the " + "requirement that the medical record must document the medical necessity of " + "furnishing the visit in the home rather than in the office." +) +P1624 = ( + "Response: We are finalizing this policy change to remove the requirement " + "that the medical record must document the medical necessity of furnishing " + "the visit in the home rather than in the office, as proposed, effective " + "January 1, 2019." +) +P3607 = ( + "As described in Medicare Claims Processing Manual, Publication 100-04, " + "Chapter 26, place of service codes are two-digit codes placed on health care " + "professional claims for CPT codes 99341 through 99350." +) + + +def _cy2019_store(tmp_path): + s = Store(":memory:", storage_dir=tmp_path / "st") + prop = s.create( + Rule( + title="Medicare Program; CY 2019 PFS Proposed Rule", + document_number="18-1", + date_published="2018-07-27", + ) + ) + final = s.create( + Rule( + title="Medicare Program; CY 2019 PFS Final Rule", + document_number="18-2", + date_published="2018-11-23", + ) + ) + s._con().executemany( + "INSERT INTO fr_anchors (item_key, p_id, page, ordinal, text) VALUES (?,?,?,?,?)", + [ + (prop, 702, 35835, 1, P702), + (final, 1622, 59630, 1, P1622), + (final, 1623, 59630, 2, P1623), + (final, 1624, 59631, 3, P1624), + (final, 3607, 59700, 4, P3607), + (final, 3608, 59700, 5, "We remove the requirement that X be done."), + (final, 3609, 59700, 6, "We are finalizing the removal of Y."), + ], + ) + s._con().execute( + "INSERT INTO fr_anchor_docs (item_key, document_number, html_url, start_page," + " end_page, fr_volume) VALUES (?,?,'u',1,2,83)", + (final, "18-2"), + ) + s._con().commit() + return s, prop, final + + +def test_coverage_rows_cite_final_rules_and_tell_a_finalized_removal(tmp_path): + s, prop, final = _cy2019_store(tmp_path) + c = _con() + sec = "100-04 ch.12 §30.6.14.1" + write_guidance( + c, + "HOME-VISITS", + [ + _g("99341", "iom", sec, src=prop, p_id=702, page=35835), + _g("99341", "iom", sec, src=final, p_id=1622, page=59630), + _g("99342", "iom", sec, src=prop, p_id=702, page=35835), + _g("99341", "iom", "100-04 ch.26", src=prop, p_id=702, page=35835), + _g("99341", "iom", "100-04 ch.26", src=final, p_id=3607, page=59700), + _g("99341", "iom", "100-04 ch.9", src=final, p_id=3608, page=59700), + _g("99350", "mln", "MLN 907166", src=prop, p_id=702, page=35835), + _g("99350", "mln", "MLN 1", src="CPT2023", p_id=0, page=0), + ], + ) + rows = guidance_origins(c, "HOME-VISITS", s) + by = {(r.code, r.reference): r for r in rows} + # proposal-cited rows are gone (99342 has only the proposal), a CPT + # edition's MLN cite stays + assert set(by) == { + ("99341", sec), + ("99341", "100-04 ch.26"), + ("99341", "100-04 ch.9"), + ("99350", "MLN 1"), + } + gone = by[("99341", sec)] + assert gone.summary == ( + "The CY2019 final rule removed the requirement in 100-04 ch.12 §30.6.14.1 " + "that the medical record must document the medical necessity of " + "furnishing the visit in the home rather than in the office, effective " + "January 1, 2019." + ) + assert (gone.kind, gone.cite_kind, gone.year, gone.extractor) == ( + "coverage", + "iom", + 2019, + "guidance", + ) + assert (gone.item_key, gone.p_id, gone.ref_p_id, gone.page_label) == ( + final, + 1624, + 1622, + "59631", + ) + assert gone.locator == f"{sec}; 83 FR 59631, ¶1624 (reference ¶1622)" + pos = by[("99341", "100-04 ch.26")] + assert (pos.item_key, pos.p_id, pos.ref_p_id) == (final, 3607, 0) + assert pos.summary == "CMS guidance for 99341 points to 100-04 ch.26." + # "remove the requirement that X" never finalized: no removal row + assert by[("99341", "100-04 ch.9")].p_id == 3608 + + +def test_removal_row_without_a_volume_year_or_effective_date(tmp_path): + from pfs.origin import _removal_row # noqa: PLC0415 + + g = _g("99341", "iom", "100-04 ch.12 §1", src="K", p_id=7) + found = {"p_id": 7, "page": 12, "req": "x " * 150, "effective": ""} + row = _removal_row(g, "100-04 ch.12 §1", ("final", None, ""), found) + # too long for 240 characters: the requirement is left out + assert row.summary == "The final rule removed a requirement in 100-04 ch.12 §1." + assert row.locator == "100-04 ch.12 §1; FR p. 12, ¶7" and row.year is None + + +# ── deletions from lineage ─────────────────────────────────────────── + +FAM_DEL = SimpleNamespace( + key="HOME-VISITS", + codes=("99324", "99339", "99341", "99343", "99350"), + synonyms=(), +) +# public FR text, CY 2023 PFS final rule (87 FR 69440 ¶390, 87 FR 69608 ¶1672) +P390 = ( + "That said, in February 2021, the AMA CPT Editorial Panel deleted the family " + "of domiciliary codes, CPT codes 99324 to 99340, and merged the services " + "described by those codes into the existing family of home-based E/M visits, " + "CPT codes 99341 to 99350 (a range of codes that includes CPT codes 99344, " + "99345, 99349, and 99350). We stated that we were not considering these " + "home-based E/M visits as potentially misvalued for CY 2023." +) +P1672 = ( + "More specifically, in its February 2021 meeting, effective beginning in 2023, " + "the CPT Editorial Panel deleted the nine CPT codes in the Domiciliary, Rest " + "Home (for example, Boarding Home), or Custodial Care Services code family " + "(CPT codes 99324-99328, and 99334-99337), and one CPT code in the Home " + "Services family (CPT code 99343), to merge these services with the eight " + "remaining home visit services. The eight remaining home services CPT codes " + "(99341, 99342, 99344, 99345, and 99347-99350) were revised to describe Home " + "or Residence Services." +) +MERGED = ( + "The AMA CPT Editorial Panel deleted domiciliary codes 99324-99340 and merged " + "them into the existing home-based E/M visit codes 99341-99350." +) + + +def _ev(code, kind, key, p_id, year=2023): + from pfs.codetables import EventRow # noqa: PLC0415 + + return EventRow(code, year, kind, "", "", key, p_id, 1, "fr", True, "") + + +def _deletion_store(tmp_path): + s, key = _chain_store( + tmp_path, + [ + (390, 69440, P390), + (1672, 69608, P1672), + (1700, 69610, "The CPT Editorial Panel deleted CPT code 99324 then."), + ( + 1701, + 69610, + "In the CY 2021 PFS final rule, the CPT Editorial Panel deleted CPT " + "code 99324, effective 2023.", + ), + ], + ) + corr = s.create( + Rule( + title="CY 2023 PFS Final Rule; Correction", + document_number="22-9", + date_published="2023-01-01", + ) + ) + s._con().execute( + "INSERT INTO fr_anchors (item_key, p_id, page, ordinal, text) VALUES (?,?,?,?,?)", + (corr, 5, 1, 5, "In 2023 the CPT Editorial Panel deleted CPT code 99324."), + ) + s._con().commit() + return s, key, corr + + +def test_deletions_come_from_lineage_and_misstated_fates_are_dropped(tmp_path): + from pfs.codetables import write_events # noqa: PLC0415 + + s, key, corr = _deletion_store(tmp_path) + c = _con() + both = ("deleted", "cpt_panel") + write_events( + c, + "99324", + [_ev("99324", k, key, p) for k in both for p in (390, 1672, 1700, 1701)] + + [_ev("99324", k, corr, 5) for k in both] + + [_ev("99324", "deleted", key, 999), _ev("99324", "cpt_panel", key, 999)] + + [_ev("99324", "telehealth_list", key, 1672)], + ) + write_events(c, "99339", [_ev("99339", k, key, 390) for k in both]) + write_events(c, "99343", [_ev("99343", k, key, 1672) for k in both]) + write_events(c, "99341", [_ev("99341", "cpt_panel", key, 1672)]) + + def ask(system, user, schema): + passage = user.split("PASSAGE:")[1] + if "That said" in passage: + return { + "origins": [ + item( + kind="cpt_panel", + reference="AMA CPT Editorial Panel", + codes=["99324", "99339", "99341", "99343", "99350"], + year=2021, + summary=MERGED, + ) + ] + } + if "More specifically" in passage: + # the model leaves out 99343 + return { + "origins": [ + item( + kind="cpt_panel", + reference="CPT Editorial Panel", + codes=["99324", "99341", "99350"], + year=2023, + summary="The panel deleted and revised codes to merge " + "the services into a single family.", + ) + ] + } + return {"origins": []} + + ask.model = "qwen-test" + rows, _ = family_origin( + c, s, FAM_DEL, ask=ask, log=RunLog(tmp_path / "l"), books=[] + ) + got = {(r.code, r.p_id, r.year, r.extractor[:7]) for r in rows} + assert got == { + # ¶390: the model row stands for the deleted codes it tells as + # deleted and for the merge targets; 99343 (deleted in this rule) + # is no merge target + ("99324", 390, 2021, "model:q"), + ("99339", 390, 2021, "model:q"), + ("99341", 390, 2021, "model:q"), + ("99350", 390, 2021, "model:q"), + # ¶1672: the deleted codes from lineage (the sentence writes 2023), + # the survivors from the model + ("99324", 1672, 2023, "lineage"), + ("99343", 1672, 2023, "lineage"), + ("99341", 1672, 2023, "model:q"), + ("99350", 1672, 2023, "model:q"), + # ¶390 is the only paragraph deleting 99339 in 2023 + ("99339", 390, 2023, "lineage"), + } + r = next(r for r in rows if (r.code, r.p_id) == ("99343", 1672)) + assert r.summary == "The CPT Editorial Panel deleted 99343, effective CY2023." + assert (r.kind, r.reference, r.cite_kind, r.item_key, r.ref_p_id) == ( + "cpt_panel", + "CPT Editorial Panel", + "fr_para", + key, + 1672, + ) + assert (r.page_label, r.locator) == ("69608", "87 FR 69608, ¶1672") + + +def test_lineage_origins_need_a_supporting_paragraph(tmp_path): + from pfs.origin import lineage_origins # noqa: PLC0415 + + s, key, corr = _deletion_store(tmp_path) + both = ("deleted", "cpt_panel") + events = { + # ¶1700 does not write the year; ¶1701 recaps another year's rule; + # the correction notice and a missing paragraph support nothing + "99324": [_ev("99324", k, key, p) for k in both for p in (1700, 1701, 999)] + + [_ev("99324", k, corr, 5) for k in both], + # deleted and cpt_panel at different paragraphs + "99343": [_ev("99343", "deleted", key, 390), _ev("99343", "cpt_panel", key, 1)], + # ¶390 names 99341 only as a merge target + "99341": [_ev("99341", k, key, 390) for k in both], + } + assert lineage_origins(s, FAM_DEL, events) == [] + + +def test_misstates_deletion(): + from pfs.origin import misstates_deletion # noqa: PLC0415 + + assert misstates_deletion(MERGED, "99343") # a merge target + assert not misstates_deletion(MERGED, "99324") + assert misstates_deletion("The Panel revised CPT code 99343.", "99343") + assert misstates_deletion("The Panel revised the home visit codes.", "99343") + assert not misstates_deletion("The Panel deleted and revised codes.", "99343") + assert not misstates_deletion("The Panel acted on the codes.", "99343") + + # ── family run ───────────────────────────────────────────────────────