feat(pfs): IOM/CFR guidance crosswalk per code family — pfs.code_guidance + stack pfs guidance (refs #689)

src/cli/pfs.py

@@ -5,6 +5,7 @@
     uv run stack pfs elements --all-payable --no-llm --dry-run
     uv run stack pfs lineage --code G2058 [--write]
     uv run stack pfs families [--write]
+    uv run stack pfs guidance --family CCM [--family …] [--write]
     uv run stack pfs review [--code G0556]
     uv run stack pfs cpt-ingest --edition 2024 [--edition 2022 …] [--dry-run]
     uv run stack pfs cpt-ingest --all
@@ -29,13 +30,16 @@ from pfs.codetables import (
     read_cpt_codes,
     read_cpt_instructions,
     read_cpt_sections,
+    read_guidance,
     write_elements,
     write_events,
     write_families,
+    write_guidance,
 )
 from pfs.cpt_load import ingest as cpt_ingest
 from pfs.extract import extract_code
 from pfs.families import FAMILIES, HAND_FAMILIES, derive_families, refresh_from
+from pfs.guidance import build as build_guidance
 from pfs.lineage import lineage
 
 log = logging.getLogger(__name__)
@@ -343,6 +347,68 @@ def families(
         con.close()
 
 
+def _print_guidance(rows: list[Any]) -> None:
+    for r in sorted(rows, key=lambda r: (r.family, r.code, r.kind, r.locator)):
+        item = r.item_key or "—"
+        typer.echo(
+            f"{r.family}  {r.kind}  {r.locator}  {item} ← {r.item_key_src} ¶{r.p_id_src}"
+        )
+
+
+@app.command()
+def guidance(
+    family: list[str] = typer.Option(
+        [], "--family", help="Family key (CCM, APCM, …); repeatable."
+    ),
+    write: bool = typer.Option(False, "--write"),
+) -> None:
+    """CFR/IOM/MLN sub-regulatory crosswalk per code family
+    (pfs.code_guidance): which CFR sections and IOM manual chapters the
+    family's FR paragraphs and the CPT manual cite, with FR provenance."""
+    if not family:
+        raise typer.BadParameter("pass --family (repeatable)")
+    keys = [f.upper() for f in family]
+    store = _store()
+    if write:
+        with _batch() as con:
+            ensure_tables(con)
+            refresh_from(con)
+            rows: list[Any] = []
+            for key in keys:
+                fam = FAMILIES.get(key)
+                if fam is None:
+                    raise typer.BadParameter(
+                        f"unknown family {key!r}; known: {', '.join(HAND_FAMILIES)}"
+                    )
+                fam_rows = build_guidance(con, store, fam.codes, family=key)
+                write_guidance(con, key, fam_rows)
+                rows.extend(fam_rows)
+        _publish()
+        _print_guidance(rows)
+        return
+    # No --write: read-only never needs the DuckDB single-writer lock a
+    # notebook may be holding (#508-#514) — read the replica (I4).
+    con = _read()
+    try:
+        refresh_from(con)
+        rows = []
+        for key in keys:
+            if key not in FAMILIES:
+                raise typer.BadParameter(
+                    f"unknown family {key!r}; known: {', '.join(HAND_FAMILIES)}"
+                )
+            try:
+                rows.extend(read_guidance(con, key))
+            except Exception as exc:
+                if _missing_table(exc):
+                    typer.echo(f"{key}: no derived guidance yet — run with --write")
+                    continue
+                raise
+        _print_guidance(rows)
+    finally:
+        con.close()
+
+
 @app.command()
 def review(code: str = typer.Option("", "--code")) -> None:
     """Element lines the classifier could not place."""

src/pfs/codetables.py

@@ -94,6 +94,28 @@ class FamilyRow:
     note: str = ""
 
 
+@dataclass(frozen=True)
+class GuidanceRow:
+    """One CFR/IOM/MLN reference a family's codes are discussed under
+    (``pfs.guidance``, #689) — ``kind`` is ``"cfr"``, ``"iom"`` or
+    ``"mln"``; ``locator`` is the canonical human-readable cite
+    ("42 CFR 410.78(a)(3)" | "100-04 ch.12 §30.6.4" | "MLN 907166");
+    ``item_key`` is the resolved bib item (``""`` when unresolved —
+    most CFR/IOM citations have no matching library item on hand);
+    ``item_key_src``/``p_id_src``/``page_src`` are the citing
+    paragraph's own provenance (an FR paragraph, or the CPT edition
+    item with ``p_id_src=0``, ``page_src=0``)."""
+
+    family: str
+    code: str
+    kind: str  # "cfr" | "iom" | "mln"
+    locator: str
+    item_key: str
+    item_key_src: str
+    p_id_src: int
+    page_src: int
+
+
 @dataclass(frozen=True)
 class CptSectionRow:
     edition_year: int
@@ -192,6 +214,9 @@ CREATE TABLE IF NOT EXISTS pfs.code_event (
 CREATE TABLE IF NOT EXISTS pfs.code_family (
     key VARCHAR, name VARCHAR, code VARCHAR, role VARCHAR, since INTEGER, until INTEGER,
     item_key VARCHAR, p_id INTEGER, note VARCHAR);
+CREATE TABLE IF NOT EXISTS pfs.code_guidance (
+    family VARCHAR, code VARCHAR, kind VARCHAR, locator VARCHAR, item_key VARCHAR,
+    item_key_src VARCHAR, p_id_src INTEGER, page_src INTEGER);
 CREATE TABLE IF NOT EXISTS pfs.cpt_section (
     edition_year INTEGER, item_key VARCHAR, sec_id VARCHAR, level INTEGER, title VARCHAR,
     path VARCHAR[], path_key VARCHAR, code_lo VARCHAR, code_hi VARCHAR, guideline VARCHAR);
@@ -321,6 +346,20 @@ def read_families(con: Any) -> list[FamilyRow]:
     return [FamilyRow(*r) for r in rows]
 
 
+def write_guidance(con: Any, family: str, rows: Sequence[GuidanceRow]) -> int:
+    ensure_tables(con)
+    con.execute("DELETE FROM pfs.code_guidance WHERE family = ?", [family])
+    return _insert(con, "pfs.code_guidance", rows)
+
+
+def read_guidance(con: Any, family: str) -> list[GuidanceRow]:
+    rows = con.execute(
+        "SELECT * FROM pfs.code_guidance WHERE family = ? ORDER BY code, kind, locator",
+        [family],
+    ).fetchall()
+    return [GuidanceRow(*r) for r in rows]
+
+
 _CPT_TABLES = (
     "pfs.cpt_section",
     "pfs.cpt_code",

src/pfs/guidance.py

@@ -0,0 +1,284 @@
+"""Sub-regulatory crosswalk: CFR sections and IOM manual chapters/sections
+a code family's rules (and the CPT manual) cite, with FR paragraph
+provenance (#689, task 6 of the 2026-09-09 code-family-anchors plan).
+
+Two sources, one extraction pass over each hit's text:
+
+1. ``harvest`` — FR paragraphs (``fr_anchors`` in the bib SQLite) that
+   name any of a family's codes. Cheap and needs no environment (the
+   bib sqlite has no ``.env`` dependency), unlike the pgvector rule
+   chunks the controller notes as an equally-valid but heavier
+   alternative source.
+2. ``harvest_cpt`` — the CPT codebook's own guideline text
+   (``pfs.cpt_section.guideline``, the heading that owns the family's
+   codes) and ``see``/``other`` instructions (``pfs.cpt_instruction``)
+   that mention Medicare policy, from the newest ingested edition.
+
+Both funnel through ``_extract``, which runs three regex families
+(CFR, IOM, MLN) over one paragraph/guideline's text and resolves each
+hit against the bib (``resolve_cfr``/``resolve_iom``) — unresolved
+references keep an empty ``item_key`` (the canonical locator is kept
+either way, Resolutions §fr_anchors). Rows are written by
+``pfs.codetables.write_guidance``/``read_guidance`` into
+``pfs.code_guidance``, deduped on ``(family, code, kind, locator,
+item_key_src, p_id_src)`` — the same citation named in the same
+paragraph for two codes in one family (a common shape: "99490 and
+99439 ... see § 410.78") legitimately yields two rows, but a citation
+matched twice by overlapping regex branches (e.g. both the titled and
+bare CFR forms firing on the same span) collapses to one.
+"""
+
+from __future__ import annotations
+
+import re
+from typing import Any, Sequence
+
+from bib.cfrlink import canonical, item_for, parse_cite
+from pfs.codetables import GuidanceRow, is_missing_table_error
+from pfs.families import FR_CITE_RE
+
+# manual display name (lower-cased, as it appears in FR prose) -> CMS
+# publication number. Controller ruling (task-6-context.md): only these
+# six IOM manuals are in scope for #689.
+_MANUAL_PUB = {
+    "claims processing": "100-04",
+    "benefit policy": "100-02",
+    "program integrity": "100-08",
+    "national coverage determinations": "100-03",
+    "managed care": "100-16",
+    "financial management": "100-06",
+}
+
+# "42 CFR 410.78(a)(3)" / "42 C.F.R. § 425.400" / "42 CFR Part 425" —
+# title-bearing form. The paragraph suffix and part-only form are left
+# to bib.cfrlink.parse_cite (fed "<title> CFR <section>") to interpret,
+# so this regex just has to isolate title/part/section/paras.
+CFR_RE = re.compile(
+    r"\b(\d{1,2})\s*C\.?F\.?R\.?\s*(?:[Pp]art\s*)?(\d{2,4})"
+    r"(?:\.(\d+[a-z]?)((?:\([^)\s]+\))*))?"
+)
+#: Bare "§ 410.78(a)(3)" — no title present. Every paragraph this module
+#: scans is a PFS rule (or the CPT manual's own Medicare-policy prose),
+#: so the title is inferred as 42 (Resolutions: kept in memory only —
+#: the stored ``locator`` is always the canonical "42 CFR ..." form,
+#: never a flag).
+BARE_CFR_RE = re.compile(r"§\s*(\d{2,4})\.(\d+[a-z]?)((?:\([^)\s]+\))*)")
+
+#: "Pub. 100-04, chapter 12" / "Publication 100-04 ... Chapter 12" (the
+#: pub-number form) or "Medicare Claims Processing Manual ... Chapter
+#: 12, Section 30.6.4" (the manual-name form) — either may open the
+#: match; both close on an optional trailing "Section N.N".
+IOM_RE = re.compile(
+    r"(?:Pub(?:lication)?\.?\s*100-0(\d)"
+    r"|(Claims Processing|Benefit Policy|Program Integrity|"
+    r"National Coverage Determinations|Managed Care|Financial Management)"
+    r"\s+Manual)"
+    r"[^.]{0,120}?[Cc]hapter\s*(\d+[A-Z]?)"
+    r"(?:[^.]{0,120}?[Ss]ection\s*([\d.]+))?"
+)
+
+#: "MLN907166" / "MLN 907166" booklet numbers (#689: optional/best-effort
+#: — no title match against bib items, just the number surfaced as a
+#: locator; always unresolved).
+MLN_RE = re.compile(r"\bMLN\s*(\d{6,7})\b", re.IGNORECASE)
+
+
+def resolve_cfr(store: Any, title: str, section: str) -> str:
+    """The bib ``regulation`` item covering "*title* CFR *section*", or
+    ``""`` when the library has no covering item (699 eCFR sections on
+    hand; most of the U.S. Code of Federal Regulations is not)."""
+    cite = parse_cite(f"{title} CFR {section}")
+    return item_for(cite, store)
+
+
+def resolve_iom(store: Any, pub: str, chapter: str) -> str:
+    """The bib item tagged ``pub:<pub>`` whose title names ``Chapter
+    <chapter>:`` (``bib.iom`` stamps chapter titles as "<manual> —
+    Chapter <N>: <title>"), or ``""`` when unresolved."""
+    marker = f"Chapter {chapter}:"
+    for item in store.list_items(tag=f"pub:{pub}"):
+        if marker in item.title:
+            return item.key
+    return ""
+
+
+def _cfr_refs(text: str) -> list[tuple[str, str]]:
+    """(title, section) pairs — *section* already shaped for
+    ``bib.cfrlink.parse_cite``'s ``"<title> CFR <section>"`` grammar
+    ("410.78(a)(3)" or "Part 425") — for every CFR reference in *text*."""
+    out: list[tuple[str, str]] = []
+    for m in CFR_RE.finditer(text):
+        title, part, dec, paras = m.group(1), m.group(2), m.group(3), m.group(4) or ""
+        section = f"{part}.{dec}{paras}" if dec else f"Part {part}"
+        out.append((title, section))
+    for m in BARE_CFR_RE.finditer(text):
+        part, dec, paras = m.group(1), m.group(2), m.group(3) or ""
+        out.append(("42", f"{part}.{dec}{paras}"))
+    return out
+
+
+def _iom_refs(text: str) -> list[tuple[str, str, str]]:
+    """(pub, chapter, section) for every IOM reference in *text* —
+    *section* is ``""`` when the sentence names a chapter but no
+    section. A manual name the pub map doesn't cover is dropped rather
+    than resolved with a guessed number."""
+    out: list[tuple[str, str, str]] = []
+    for m in IOM_RE.finditer(text):
+        pub_digit, manual_name, chapter, section = m.groups()
+        pub = (
+            f"100-0{pub_digit}"
+            if pub_digit
+            else _MANUAL_PUB.get((manual_name or "").lower(), "")
+        )
+        if not pub:
+            continue
+        # A sentence-final "Section 10.2.5.2." swallows the closing
+        # period into [\d.]+ — trim it; a real section number never
+        # ends in ".".
+        out.append((pub, chapter.upper(), (section or "").rstrip(".")))
+    return out
+
+
+def _extract(text: str, store: Any) -> list[tuple[str, str, str]]:
+    """(kind, locator, item_key) for every CFR/IOM/MLN reference in
+    *text*, resolved against *store* — ``item_key`` is ``""`` when
+    unresolved."""
+    out: list[tuple[str, str, str]] = []
+    for title, section in _cfr_refs(text):
+        cite = parse_cite(f"{title} CFR {section}")
+        out.append(("cfr", canonical(cite), resolve_cfr(store, title, section)))
+    for pub, chapter, section in _iom_refs(text):
+        locator = f"{pub} ch.{chapter}" + (f" §{section}" if section else "")
+        out.append(("iom", locator, resolve_iom(store, pub, chapter)))
+    for m in MLN_RE.finditer(text):
+        out.append(("mln", f"MLN {m.group(1)}", ""))
+    return out
+
+
+def _code_pattern(codes: Sequence[str]) -> re.Pattern[str]:
+    alts = "|".join(re.escape(c.upper()) for c in codes)
+    return re.compile(rf"\b(?:{alts})\b")
+
+
+def _codes_in(text: str, pattern: re.Pattern[str]) -> tuple[str, ...]:
+    """Family codes literally present in *text*, with FR citations
+    ("91 FR 43842") stripped first — a citation's page number reads as
+    a CPT code and must never fire a false match (mirrors
+    ``pfs.families.find_codes``)."""
+    stripped = FR_CITE_RE.sub(" ", text)
+    return tuple(sorted({m.group(0) for m in pattern.finditer(stripped)}))
+
+
+def dedupe(rows: Sequence[GuidanceRow]) -> list[GuidanceRow]:
+    """First-seen-wins de-duplication on ``(family, code, kind,
+    locator, item_key_src, p_id_src)`` — see the module docstring."""
+    seen: set[tuple[str, str, str, str, str, int]] = set()
+    out: list[GuidanceRow] = []
+    for r in rows:
+        key = (r.family, r.code, r.kind, r.locator, r.item_key_src, r.p_id_src)
+        if key in seen:
+            continue
+        seen.add(key)
+        out.append(r)
+    return out
+
+
+def harvest(store: Any, codes: Sequence[str], *, family: str) -> list[GuidanceRow]:
+    """FR paragraphs (``fr_anchors``) naming any of *codes* -> the
+    CFR/IOM/MLN references they cite, anchored to that paragraph
+    (``item_key_src``/``p_id_src``/``page_src``)."""
+    if not codes:
+        return []
+    con = store._con()  # noqa: SLF001 — same pattern as pfs.descriptors
+    pattern = _code_pattern(codes)
+    like_clause = " OR ".join("text LIKE ?" for _ in codes)
+    params = [f"%{c.upper()}%" for c in codes]
+    rows = con.execute(
+        f"SELECT item_key, p_id, page, text FROM fr_anchors WHERE {like_clause}",
+        params,
+    ).fetchall()
+    out: list[GuidanceRow] = []
+    for item_key, p_id, page, text in rows:
+        present = _codes_in(text, pattern)
+        if not present:
+            continue
+        for kind, locator, resolved in _extract(text, store):
+            for code in present:
+                out.append(
+                    GuidanceRow(
+                        family, code, kind, locator, resolved, item_key, p_id, page
+                    )
+                )
+    return dedupe(out)
+
+
+def harvest_cpt(
+    con: Any, store: Any, codes: Sequence[str], *, family: str
+) -> list[GuidanceRow]:
+    """The newest ingested CPT edition's guideline text and ``see``/
+    ``other`` instructions for *codes* -> the CFR/IOM/MLN references
+    they cite, anchored to the CPT edition item (``p_id_src=0``,
+    ``page_src=0`` — no FR paragraph; locating a page inside the PDF
+    chapter is out of scope, #689). ``[]`` on a replica with no
+    ``pfs.cpt_*`` tables yet (I4: not a bug, just nothing ingested)."""
+    from pfs.codetables import (
+        cpt_years,
+        read_cpt_codes,
+        read_cpt_instructions,
+        read_cpt_sections,
+    )
+
+    if not codes:
+        return []
+    try:
+        years = cpt_years(con)
+    except Exception as exc:
+        if is_missing_table_error(exc):
+            return []
+        raise
+    if not years:
+        return []
+    year = max(years)
+    codes_u = {c.upper() for c in codes}
+    sections = {s.sec_id: s for s in read_cpt_sections(con, year)}
+
+    codes_by_sec: dict[str, set[str]] = {}
+    for c in read_cpt_codes(con, year):
+        if c.code in codes_u:
+            codes_by_sec.setdefault(c.sec_id, set()).add(c.code)
+
+    out: list[GuidanceRow] = []
+    for sec_id, fam_codes in codes_by_sec.items():
+        section = sections.get(sec_id)
+        if section is None or not section.guideline:
+            continue
+        for kind, locator, resolved in _extract(section.guideline, store):
+            for code in sorted(fam_codes):
+                out.append(
+                    GuidanceRow(
+                        family, code, kind, locator, resolved, section.item_key, 0, 0
+                    )
+                )
+
+    for instr in read_cpt_instructions(con, year):
+        if instr.code not in codes_u or instr.kind not in ("see", "other"):
+            continue
+        if not any(k in instr.text for k in ("CFR", "Medicare", "Chapter", "chapter")):
+            continue
+        for kind, locator, resolved in _extract(instr.text, store):
+            out.append(
+                GuidanceRow(
+                    family, instr.code, kind, locator, resolved, instr.item_key, 0, 0
+                )
+            )
+    return dedupe(out)
+
+
+def build(
+    con: Any, store: Any, codes: Sequence[str], *, family: str
+) -> list[GuidanceRow]:
+    """Every guidance row for one family — FR (``harvest``) + CPT
+    (``harvest_cpt``), deduped across both sources."""
+    rows = harvest(store, codes, family=family)
+    rows.extend(harvest_cpt(con, store, codes, family=family))
+    return dedupe(rows)

tests/cli/test_pfs_cli.py

@@ -1,4 +1,4 @@
-"""stack pfs — elements / lineage / families / review."""
+"""stack pfs — elements / lineage / families / guidance / review."""
 
 from __future__ import annotations
 
@@ -11,13 +11,16 @@ from cli import app
 from pfs.codetables import (
     ElementRow,
     EventRow,
+    GuidanceRow,
     ensure_tables,
     read_elements,
     read_events,
     read_families,
+    read_guidance,
     write_cpt_edition,
     write_elements,
     write_events,
+    write_guidance,
 )
 from pfs.cpt_model import CptCode, CptEdition, CptSection
 from pfs.extract import Extraction
@@ -534,6 +537,90 @@ class TestCptIngest:
         assert con.published == [True]
 
 
+class TestGuidance:
+    def _row(self, family="CCM", code="99490"):
+        return GuidanceRow(
+            family, code, "cfr", "42 CFR 410.78(a)(3)", "SEC", "K", 398, 32389
+        )
+
+    def test_write_builds_writes_and_publishes(self, con, monkeypatch):
+        seen = {}
+
+        def fake_build(con_, store_, codes, *, family):
+            seen["con"] = con_
+            seen["store"] = store_
+            seen["codes"] = codes
+            seen["family"] = family
+            return [self._row(family=family)]
+
+        monkeypatch.setattr(pfs_cli, "build_guidance", fake_build)
+        res = runner.invoke(app, ["pfs", "guidance", "--family", "ccm", "--write"])
+        assert res.exit_code == 0, res.output
+        assert seen["family"] == "CCM"
+        assert seen["con"] is not None
+        assert seen["store"] is not None
+        assert set(seen["codes"]) == set(pfs_cli.HAND_FAMILIES["CCM"].codes)
+        got = read_guidance(con, "CCM")
+        assert [r.code for r in got] == ["99490"]
+        assert con.published == [True]
+        assert "CCM  cfr  42 CFR 410.78(a)(3)  SEC ← K ¶398" in res.output
+
+    def test_write_scopes_per_family(self, con, monkeypatch):
+        monkeypatch.setattr(
+            pfs_cli,
+            "build_guidance",
+            lambda con_, store_, codes, *, family: [self._row(family=family)],
+        )
+        res = runner.invoke(
+            app, ["pfs", "guidance", "--family", "CCM", "--family", "APCM", "--write"]
+        )
+        assert res.exit_code == 0, res.output
+        assert [r.family for r in read_guidance(con, "CCM")] == ["CCM"]
+        assert [r.family for r in read_guidance(con, "APCM")] == ["APCM"]
+
+    def test_without_write_reads_and_never_takes_the_write_lock(self, con, monkeypatch):
+        write_guidance(con, "CCM", [self._row()])
+
+        def fail_batch():
+            raise AssertionError("must not open a RW duckdb_batch connection")
+
+        monkeypatch.setattr(pfs_cli, "_batch", fail_batch)
+        res = runner.invoke(app, ["pfs", "guidance", "--family", "CCM"])
+        assert res.exit_code == 0, res.output
+        assert "CCM  cfr  42 CFR 410.78(a)(3)  SEC ← K ¶398" in res.output
+        assert con.published == []
+
+    def test_no_derived_guidance_yet_is_a_friendly_no_op(self, monkeypatch):
+        # I4: a fresh replica with no pfs.code_guidance (guidance --write
+        # hasn't run yet) is not a bug — print a friendly message per
+        # family and exit 0, without ever taking the write lock.
+        bare = duckdb.connect(":memory:")
+        monkeypatch.setattr(pfs_cli, "_read", lambda: bare)
+
+        def fail_batch():
+            raise AssertionError("must not open a RW duckdb_batch connection")
+
+        monkeypatch.setattr(pfs_cli, "_batch", fail_batch)
+        try:
+            res = runner.invoke(app, ["pfs", "guidance", "--family", "CCM"])
+            assert res.exit_code == 0, res.output
+            assert "no derived guidance yet" in res.output
+        finally:
+            bare.close()
+
+    def test_unknown_family_error_lists_hand_families_only(self, con, monkeypatch):
+        pfs_cli.FAMILIES["ZZZZ-99999"] = pfs_cli.FAMILIES["CCM"]
+        res = runner.invoke(app, ["pfs", "guidance", "--family", "NOPE", "--write"])
+        assert res.exit_code != 0
+        assert "ZZZZ-99999" not in res.output
+        for key in pfs_cli.HAND_FAMILIES:
+            assert key in res.output
+
+    def test_requires_family(self, con):
+        res = runner.invoke(app, ["pfs", "guidance"])
+        assert res.exit_code != 0
+
+
 class TestReview:
     def test_lists_queue(self, con):
         from pfs.codetables import ReviewRow

tests/pfs/test_guidance.py

@@ -0,0 +1,313 @@
+"""pfs.guidance — sub-regulatory crosswalk (CFR/IOM/MLN) per code family
+(#689, task 6). Regex snippets are real ``fr_anchors`` text (see
+task-6-report.md for item_key/p_id provenance)."""
+
+from __future__ import annotations
+
+import duckdb
+import pytest
+
+from bib.item import Manual, Regulation, Rule
+from bib.store import Store
+from pfs.codetables import (
+    GuidanceRow,
+    ensure_tables,
+    read_guidance,
+    write_guidance,
+)
+from pfs.guidance import (
+    BARE_CFR_RE,
+    CFR_RE,
+    IOM_RE,
+    MLN_RE,
+    _cfr_refs,
+    build,
+    dedupe,
+    harvest,
+    harvest_cpt,
+    resolve_cfr,
+    resolve_iom,
+)
+
+# ── real fr_anchors snippets ──────────────────────────────────────────
+
+BARE_CFR_TEXT = (
+    "Step 3. Review the elements of the service as described by the "
+    "HCPCS code and determine whether each of them is capable of being "
+    "furnished using an interactive telecommunications system as "
+    "defined in § 410.78(a)(3)."
+)  # 2KVJ2HKX ¶398
+
+IOM_MANUAL_TEXT = (
+    "Consistent with the established Medicare HPSA physician bonus "
+    "program (Medicare Claims Processing Manual, Pub. 100-04, Chapter "
+    "12, Section 90.4.4) and the proposed Health Professional Shortage "
+    "Area Surgical Incentive Payment Program (HSIP) described in "
+    "section III.S.2. of this proposed rule, we are proposing that "
+    "PCIP payments would be calculated by the Medicare contractors."
+)  # T4N6JINW ¶1502
+
+IOM_PUB_TEXT = (
+    "A physician may not bill Medicare for a service that is on the "
+    "list of “always therapy” services (see Pub. 100-04, the "
+    "Medicare Benefit Policy Manual, chapter 5, section 20) if the "
+    "service was done by staff that is not qualified to provide a "
+    "skilled therapy service."
+)  # WBS77GY5 ¶1771
+
+
+# ── regexes ─────────────────────────────────────────────────────────
+
+
+class TestCfrRegex:
+    def test_bare_section_infers_no_title(self):
+        # BARE_CFR_RE never captures a title — resolve_cfr's caller
+        # supplies the inferred "42" (Resolutions: title_inferred kept
+        # only in memory, never as a stored flag).
+        m = BARE_CFR_RE.search(BARE_CFR_TEXT)
+        assert m is not None
+        assert m.groups() == ("410", "78", "(a)(3)")
+
+    def test_titled_cfr_captures_title_part_section_paras(self):
+        m = CFR_RE.search("under 42 CFR 410.78(a)(3) and elsewhere")
+        assert m is not None
+        assert m.groups() == ("42", "410", "78", "(a)(3)")
+
+    def test_titled_cfr_part_only(self):
+        m = CFR_RE.search("see 42 CFR Part 425 for details")
+        assert m is not None
+        assert m.groups() == ("42", "425", None, None)
+
+    def test_dotted_cfr_with_section_sign_resolves_via_bare_fallback(self):
+        # CFR_RE (titled) has no "§" between "C.F.R." and the number —
+        # the bare regex's title-inferred fallback picks it up instead,
+        # landing on the same (title, section) either way.
+        assert CFR_RE.search("42 C.F.R. § 414.1425 governs this") is None
+        assert _cfr_refs("42 C.F.R. § 414.1425 governs this") == [("42", "414.1425")]
+
+
+class TestIomRegex:
+    def test_manual_name_form(self):
+        m = IOM_RE.search(IOM_MANUAL_TEXT)
+        assert m is not None
+        pub_digit, manual_name, chapter, section = m.groups()
+        # The pub-number branch fires (it starts later in the string but
+        # is the only branch the Manual-name gap can't cross — "Pub."
+        # has a literal period the [^.]{0,120}? gap can't skip past).
+        assert pub_digit == "4"
+        assert manual_name is None
+        assert chapter == "12"
+        assert section == "90.4.4"
+
+    def test_pub_number_form(self):
+        m = IOM_RE.search(IOM_PUB_TEXT)
+        assert m is not None
+        pub_digit, manual_name, chapter, section = m.groups()
+        assert pub_digit == "4"
+        assert chapter == "5"
+        assert section == "20"
+
+    def test_no_match_without_chapter(self):
+        assert IOM_RE.search("Pub. 100-04 discusses billing generally.") is None
+
+
+class TestMlnRegex:
+    def test_matches_booklet_number(self):
+        m = MLN_RE.search("see MLN907166 for details")
+        assert m is not None
+        assert m.group(1) == "907166"
+
+    def test_case_insensitive_with_space(self):
+        assert MLN_RE.search("mln 1234567").group(1) == "1234567"
+
+
+# ── resolvers ───────────────────────────────────────────────────────
+
+
+@pytest.fixture
+def store():
+    s = Store(":memory:")
+    s.create(
+        Regulation(
+            title="42 CFR Part 410",
+            url="https://www.ecfr.gov/current/title-42/part-410",
+        )
+    )
+    sec_key = s.create(
+        Regulation(
+            title="42 CFR 410.78",
+            url="https://www.ecfr.gov/current/title-42/section-410.78",
+        )
+    )
+    manual_key = s.create(
+        Manual(
+            title=(
+                "Medicare Claims Processing Manual — Chapter 12: "
+                "Physicians/Nonphysician Practitioners"
+            )
+        )
+    )
+    s.add_tag(manual_key, "pub:100-04")
+    yield s, sec_key, manual_key
+    s.close()
+
+
+class TestResolveCfr:
+    def test_resolves_exact_section(self, store):
+        s, sec_key, _ = store
+        assert resolve_cfr(s, "42", "410.78(a)(3)") == sec_key
+
+    def test_unresolved_returns_empty(self, store):
+        s, _, _ = store
+        assert resolve_cfr(s, "45", "155.20") == ""
+
+
+class TestResolveIom:
+    def test_resolves_chapter_by_pub_and_title(self, store):
+        s, _, manual_key = store
+        assert resolve_iom(s, "100-04", "12") == manual_key
+
+    def test_unresolved_pub_returns_empty(self, store):
+        s, _, _ = store
+        assert resolve_iom(s, "100-02", "12") == ""
+
+    def test_unresolved_chapter_returns_empty(self, store):
+        s, _, _ = store
+        assert resolve_iom(s, "100-04", "5") == ""
+
+
+# ── harvest (FR paragraphs) ────────────────────────────────────────
+
+
+@pytest.fixture
+def store_with_anchors(store):
+    s, sec_key, manual_key = store
+    con = s._con()  # noqa: SLF001
+    item_key = s.create(Rule(title="CY2020 PFS final rule"))
+    con.executemany(
+        "INSERT INTO fr_anchors (item_key, p_id, page, ordinal, text) VALUES (?,?,?,?,?)",
+        [
+            (item_key, 398, 32389, 1, f"For code 99490, {BARE_CFR_TEXT}"),
+            (item_key, 1502, 100, 2, f"Codes 99490 and 99439: {IOM_MANUAL_TEXT}"),
+            (item_key, 9, 5, 3, "no codes named here at all"),
+        ],
+    )
+    con.commit()
+    yield s, item_key, sec_key, manual_key
+    s.close()
+
+
+class TestHarvest:
+    def test_bare_cfr_paragraph_resolves(self, store_with_anchors):
+        s, item_key, sec_key, _manual_key = store_with_anchors
+        rows = harvest(s, ("99490", "99439"), family="CCM")
+        cfr_rows = [r for r in rows if r.kind == "cfr"]
+        assert len(cfr_rows) == 1
+        r = cfr_rows[0]
+        assert r.family == "CCM"
+        assert r.code == "99490"
+        assert r.locator == "42 CFR 410.78(a)(3)"
+        assert r.item_key == sec_key
+        assert r.item_key_src == item_key
+        assert r.p_id_src == 398
+        assert r.page_src == 32389
+
+    def test_iom_paragraph_names_both_codes(self, store_with_anchors):
+        s, item_key, _sec_key, manual_key = store_with_anchors
+        rows = harvest(s, ("99490", "99439"), family="CCM")
+        iom_rows = [r for r in rows if r.kind == "iom"]
+        codes = {r.code for r in iom_rows}
+        assert codes == {"99490", "99439"}
+        assert all(r.locator == "100-04 ch.12 §90.4.4" for r in iom_rows)
+        assert all(r.item_key == manual_key for r in iom_rows)
+
+    def test_paragraph_without_codes_is_skipped(self, store_with_anchors):
+        s, _item_key, _sec_key, _manual_key = store_with_anchors
+        rows = harvest(s, ("99490", "99439"), family="CCM")
+        assert all(r.p_id_src != 9 for r in rows)
+
+    def test_empty_codes_returns_empty(self, store_with_anchors):
+        s, *_rest = store_with_anchors
+        assert harvest(s, (), family="CCM") == []
+
+    def test_fr_citation_page_number_never_fires_as_a_code(self, store_with_anchors):
+        # A code equal to an FR citation's page number ("43842") must
+        # not be treated as present just because "91 FR 43842" appears.
+        s, item_key, _sec_key, _manual_key = store_with_anchors
+        con = s._con()  # noqa: SLF001
+        con.execute(
+            "INSERT INTO fr_anchors (item_key, p_id, page, ordinal, text) VALUES (?,?,?,?,?)",
+            (item_key, 500, 1, 4, "See 91 FR 43842 for background."),
+        )
+        con.commit()
+        rows = harvest(s, ("43842",), family="FAKE")
+        assert rows == []
+
+
+# ── dedupe ──────────────────────────────────────────────────────────
+
+
+class TestDedupe:
+    def test_collapses_identical_keys(self):
+        r = GuidanceRow("CCM", "99490", "cfr", "42 CFR 410.78", "K", "SRC", 1, 2)
+        assert dedupe([r, r]) == [r]
+
+    def test_keeps_distinct_codes(self):
+        r1 = GuidanceRow("CCM", "99490", "cfr", "42 CFR 410.78", "K", "SRC", 1, 2)
+        r2 = GuidanceRow("CCM", "99439", "cfr", "42 CFR 410.78", "K", "SRC", 1, 2)
+        assert dedupe([r1, r2]) == [r1, r2]
+
+
+# ── DDL / write / read round trip ────────────────────────────────────
+
+
+@pytest.fixture
+def con():
+    c = duckdb.connect(":memory:")
+    ensure_tables(c)
+    yield c
+    c.close()
+
+
+class TestGuidanceTable:
+    def test_write_then_read_round_trips(self, con):
+        rows = [
+            GuidanceRow(
+                "CCM", "99490", "cfr", "42 CFR 410.78(a)(3)", "SEC", "K", 398, 32389
+            ),
+            GuidanceRow(
+                "CCM", "99439", "iom", "100-04 ch.12 §90.4.4", "", "K", 1502, 100
+            ),
+        ]
+        n = write_guidance(con, "CCM", rows)
+        assert n == 2
+        got = read_guidance(con, "CCM")
+        assert {(r.code, r.kind) for r in got} == {("99490", "cfr"), ("99439", "iom")}
+
+    def test_write_is_delete_then_insert_per_family(self, con):
+        write_guidance(
+            con, "CCM", [GuidanceRow("CCM", "99490", "cfr", "x", "", "K", 1, 1)]
+        )
+        write_guidance(
+            con, "CCM", [GuidanceRow("CCM", "99439", "cfr", "y", "", "K", 2, 2)]
+        )
+        got = read_guidance(con, "CCM")
+        assert [r.code for r in got] == ["99439"]
+
+    def test_other_family_untouched(self, con):
+        write_guidance(
+            con, "APCM", [GuidanceRow("APCM", "G0556", "cfr", "x", "", "K", 1, 1)]
+        )
+        write_guidance(con, "CCM", [])
+        assert len(read_guidance(con, "APCM")) == 1
+
+
+# ── build (FR + CPT, no CPT tables ingested) ─────────────────────────
+
+
+class TestBuild:
+    def test_build_tolerates_missing_cpt_tables(self, store_with_anchors, con):
+        s, *_rest = store_with_anchors
+        rows = build(con, s, ("99490", "99439"), family="CCM")
+        assert rows  # the FR-sourced rows still come through
+        assert harvest_cpt(con, s, ("99490",), family="CCM") == []