merge: P49 slice 1 — code families as first-class objects: element model, lineage anchors, derived families, stack pfs CLI (refs #684 #685 #686 #687)

Merges PR #701 (worktree-p49-code-family-foundation). Spec docs/superpowers/specs/2026-09-09-code-family-longitudinal-design.md; plan docs/superpowers/plans/2026-09-09-code-family-foundation.md. Follow-ups #698 #699 #700.

Claude-Session: https://claude.ai/code/session_01Aum3pEMAM3yQVdFSdVe6Gc

docs/docs/cli/pfs-elements.md

@@ -0,0 +1,23 @@
+---
+title: stack pfs elements
+sidebar_position: 63
+---
+
+# `stack pfs elements`
+
+```
+Usage: stack pfs elements [OPTIONS]                                            
+                                                                                
+ Extract typed elements for codes into pfs.code_element (+ review queue).       
+                                                                                
+╭─ Options ────────────────────────────────────────────────────────────────────╮
+│ --code               TEXT  HCPCS/CPT code (repeatable).                      │
+│ --family             TEXT  Expand a registered family (CCM, APCM, …);        │
+│                            repeatable.                                       │
+│ --all-payable              Every A/R/T code in the newest RVU year.          │
+│ --no-llm                   Skip the local-model classifier (unknown lines go │
+│                            to review).                                       │
+│ --dry-run                  Extract and report; write nothing.                │
+│ --help                     Show this message and exit.                       │
+╰──────────────────────────────────────────────────────────────────────────────╯
+```

docs/docs/cli/pfs-families.md

@@ -0,0 +1,17 @@
+---
+title: stack pfs families
+sidebar_position: 65
+---
+
+# `stack pfs families`
+
+```
+Usage: stack pfs families [OPTIONS]                                            
+                                                                                
+ Derive families from pfs.code_element / pfs.code_event / pfs.rvu.              
+                                                                                
+╭─ Options ────────────────────────────────────────────────────────────────────╮
+│ --write                                                                      │
+│ --help           Show this message and exit.                                 │
+╰──────────────────────────────────────────────────────────────────────────────╯
+```

docs/docs/cli/pfs-lineage.md

@@ -0,0 +1,18 @@
+---
+title: stack pfs lineage
+sidebar_position: 64
+---
+
+# `stack pfs lineage`
+
+```
+Usage: stack pfs lineage [OPTIONS]                                             
+                                                                                
+ Timeline of a code: RVU-file diffs and FR paragraphs, cross-checked.           
+                                                                                
+╭─ Options ────────────────────────────────────────────────────────────────────╮
+│ *  --code         TEXT  [required]                                           │
+│    --write              Persist to pfs.code_event and republish.             │
+│    --help               Show this message and exit.                          │
+╰──────────────────────────────────────────────────────────────────────────────╯
+```

docs/docs/cli/pfs-review.md

@@ -0,0 +1,17 @@
+---
+title: stack pfs review
+sidebar_position: 66
+---
+
+# `stack pfs review`
+
+```
+Usage: stack pfs review [OPTIONS]                                              
+                                                                                
+ Element lines the classifier could not place.                                  
+                                                                                
+╭─ Options ────────────────────────────────────────────────────────────────────╮
+│ --code        TEXT                                                           │
+│ --help              Show this message and exit.                              │
+╰──────────────────────────────────────────────────────────────────────────────╯
+```

docs/docs/cli/pfs.md

@@ -0,0 +1,24 @@
+---
+title: stack pfs
+sidebar_position: 62
+---
+
+# `stack pfs`
+
+```
+Usage: stack pfs [OPTIONS] COMMAND [ARGS]...                                   
+                                                                                
+ Code elements, lineage events and derived families.                            
+                                                                                
+╭─ Options ────────────────────────────────────────────────────────────────────╮
+│ --help          Show this message and exit.                                  │
+╰──────────────────────────────────────────────────────────────────────────────╯
+╭─ Commands ───────────────────────────────────────────────────────────────────╮
+│ elements  Extract typed elements for codes into pfs.code_element (+ review   │
+│           queue).                                                            │
+│ lineage   Timeline of a code: RVU-file diffs and FR paragraphs,              │
+│           cross-checked.                                                     │
+│ families  Derive families from pfs.code_element / pfs.code_event / pfs.rvu.  │
+│ review    Element lines the classifier could not place.                      │
+╰──────────────────────────────────────────────────────────────────────────────╯
+```

docs/docs/cli/stack.md

@@ -23,9 +23,17 @@ Usage: stack [OPTIONS] COMMAND [ARGS]...
 │ load      Ingest data (CCLF, BCDA, seeds).                                   │
 │ generate  Regenerate code artefacts.                                         │
 │ bib       Bibliography operations.                                           │
+│ comments  CMS rulemaking comment text extraction & analysis.                 │
 │ lake      Lakehouse schema and data.                                         │
+│ llm       Local RAG — index, search, tag.                                    │
 │ db        DuckDB utilities.                                                  │
 │ docs      Documentation generation.                                          │
 │ api       API server.                                                        │
+│ perf      Pipeline telemetry utilities.                                      │
+│ rec       Reconcile calculated payments against CMS ground-truth files.      │
+│ pfs       Code elements, lineage events and derived families.                │
+│ zot       Zotero database maintenance.                                       │
+│ prisma    LLM-driven PRISMA screening workflow.                              │
+│ mail      Maddy mail server (DO + Resend smarthost).                         │
 ╰──────────────────────────────────────────────────────────────────────────────╯
 ```

src/cli/__init__.py

@@ -24,6 +24,7 @@ from cli.llm import app as llm_app
 from cli.load import app as load_app
 from cli.mail import app as mail_app
 from cli.perf import app as perf_app
+from cli.pfs import app as pfs_app
 from cli.prisma import app as prisma_app
 from cli.rec import app as rec_app
 from cli.run import run
@@ -58,6 +59,9 @@ app.add_typer(
     name="rec",
     help="Reconcile calculated payments against CMS ground-truth files.",
 )
+app.add_typer(
+    pfs_app, name="pfs", help="Code elements, lineage events and derived families."
+)
 app.add_typer(zot_app, name="zot", help="Zotero database maintenance.")
 app.add_typer(prisma_app, name="prisma", help="LLM-driven PRISMA screening workflow.")
 app.add_typer(mail_app, name="mail", help="Maddy mail server (DO + Resend smarthost).")

src/cli/pfs.py

@@ -0,0 +1,280 @@
+"""stack pfs — code elements, lineage events and derived families.
+
+    uv run stack pfs elements --code G0556 --code 99490
+    uv run stack pfs elements --family CCM
+    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 review [--code G0556]
+
+Writes go through ``duckdb_batch`` (single-writer rule) and republish
+the read-only replica so the chat and notebooks see them.
+"""
+
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+import typer
+
+from pfs.codetables import (
+    ensure_tables,
+    read_all_elements,
+    read_all_events,
+    write_elements,
+    write_events,
+    write_families,
+)
+from pfs.extract import extract_code
+from pfs.families import FAMILIES, HAND_FAMILIES, derive_families, refresh_from
+from pfs.lineage import lineage
+
+log = logging.getLogger(__name__)
+
+app = typer.Typer(no_args_is_help=True)
+
+
+#: A replica that hasn't had `elements`/`lineage --write` (or an initial
+#: `families --write`) run yet is missing `pfs.code_element`/
+#: `pfs.code_event`/`pfs.code_family` — not a bug, just nothing derived
+#: yet (I4). DuckDB's "table does not exist" errors are Catalog errors.
+def _missing_table(exc: Exception) -> bool:
+    msg = str(exc)
+    return "Catalog" in msg or "does not exist" in msg
+
+
+# ── indirections the tests monkeypatch ───────────────────────────────
+def _batch() -> Any:
+    from conf.connect import duckdb_batch
+
+    return duckdb_batch("aco")
+
+
+def _read() -> Any:
+    from conf.connect import duckdb
+
+    return duckdb("aco", read_only=True)
+
+
+def _store() -> Any:
+    from conf.connect import bib
+
+    return bib()
+
+
+def _publish() -> None:
+    from conf.connect import publish_replica
+
+    typer.echo(f"replica → {publish_replica('aco')}")
+
+
+def _classifier() -> Any:
+    from llm import config as llm_config
+    from llm.classify import closed_vocab_classifier
+    from llm.pool import HostPool
+
+    cfg = llm_config.load()
+    return closed_vocab_classifier(cfg, HostPool.from_config(cfg))
+
+
+def _run_elements(
+    con: Any, store: Any, targets: list[str], classify: Any, *, write: bool
+) -> None:
+    for c in targets:
+        x = extract_code(store, con, c, classify=classify)
+        if write:
+            write_elements(con, c, x.rows, x.reviews)
+        typer.echo(f"{c}: {len(x.rows)} elements, {len(x.reviews)} for review")
+
+
+def _codes_for(
+    con: Any, codes: list[str], families: list[str], all_payable: bool
+) -> list[str]:
+    out: list[str] = [c.upper() for c in codes]
+    if families:
+        refresh_from(con)
+        for family in families:
+            fam = FAMILIES.get(family.upper())
+            if fam is None:
+                # List only the hand families here (M3) — FAMILIES can hold
+                # ~15k derived keys once `refresh_from` has run, and that
+                # dump is useless in an error message.
+                raise typer.BadParameter(
+                    f"unknown family {family!r}; known: {', '.join(HAND_FAMILIES)}"
+                )
+            out.extend(fam.codes)
+    if all_payable:
+        rows = con.execute(
+            "SELECT DISTINCT hcpcs FROM pfs.rvu WHERE status_code IN ('A','R','T') "
+            "AND year = (SELECT max(year) FROM pfs.rvu) ORDER BY hcpcs"
+        ).fetchall()
+        log.warning(
+            "--all-payable is experimental: targeting %d codes (~20 min of SQL "
+            "before any model call)",
+            len(rows),
+        )
+        out.extend(r[0] for r in rows)
+    if not out:
+        raise typer.BadParameter("pass --code, --family or --all-payable")
+    return sorted(set(out))
+
+
+@app.command()
+def elements(
+    code: list[str] = typer.Option([], "--code", help="HCPCS/CPT code (repeatable)."),
+    family: list[str] = typer.Option(
+        [], "--family", help="Expand a registered family (CCM, APCM, …); repeatable."
+    ),
+    all_payable: bool = typer.Option(
+        False,
+        "--all-payable",
+        help=(
+            "Every A/R/T code in the newest RVU year (experimental: ~20 min "
+            "of SQL before any model call)."
+        ),
+    ),
+    no_llm: bool = typer.Option(
+        False,
+        "--no-llm",
+        help="Skip the local-model classifier (unknown lines go to review).",
+    ),
+    dry_run: bool = typer.Option(
+        False, "--dry-run", help="Extract and report; write nothing."
+    ),
+) -> None:
+    """Extract typed elements for codes into pfs.code_element (+ review queue)."""
+    store = _store()
+    classify = None if no_llm else _classifier()
+    if dry_run:
+        # A preview never contends for the DuckDB single-writer lock a
+        # notebook may be holding (#508-#514) — read the replica, don't
+        # open a batch writer.
+        con = _read()
+        try:
+            targets = _codes_for(con, code, family, all_payable)
+            _run_elements(con, store, targets, classify, write=False)
+        finally:
+            con.close()
+    else:
+        with _batch() as con:
+            ensure_tables(con)
+            targets = _codes_for(con, code, family, all_payable)
+            _run_elements(con, store, targets, classify, write=True)
+        _publish()
+
+
+def _print_events(events: list[Any]) -> None:
+    for e in events:
+        arrow = f"{e.from_codes or '·'}→{e.to_codes or '·'}"
+        anchor = f"{e.item_key} ¶{e.p_id}" if e.item_key else f"rvu {e.note}".strip()
+        flag = "" if e.anchored else "  UNANCHORED"
+        typer.echo(f"{e.year}  {e.kind:<18}{arrow:<16}{anchor}{flag}")
+
+
+@app.command("lineage")
+def lineage_cmd(
+    code: str = typer.Option(..., "--code"),
+    write: bool = typer.Option(
+        False, "--write", help="Persist to pfs.code_event and republish."
+    ),
+) -> None:
+    """Timeline of a code: RVU-file diffs and FR paragraphs, cross-checked."""
+    store = _store()
+    if write:
+        with _batch() as con:
+            ensure_tables(con)
+            events = lineage(con, store, code.upper())
+            _print_events(events)
+            write_events(con, code.upper(), events)
+        _publish()
+        return
+    # No --write: a plain lookup never needs the DuckDB single-writer lock
+    # a notebook may be holding (#508-#514) — read the replica (I4).
+    con = _read()
+    try:
+        try:
+            events = lineage(con, store, code.upper())
+        except Exception as exc:
+            if _missing_table(exc):
+                typer.echo("no derived tables yet — run with --write")
+                return
+            raise
+        _print_events(events)
+    finally:
+        con.close()
+
+
+def _derive_rows(con: Any) -> list[Any]:
+    codes = [
+        r[0]
+        for r in con.execute(
+            "SELECT DISTINCT code FROM pfs.code_element UNION SELECT DISTINCT code FROM pfs.code_event "
+            "UNION SELECT DISTINCT hcpcs FROM pfs.rvu WHERE year = (SELECT max(year) FROM pfs.rvu) AND status_code IN ('A','R','T')"
+        ).fetchall()
+    ]
+    # Two full-table reads grouped in Python instead of ~2xN per-code
+    # queries run inside the write lock (I9).
+    all_elements = read_all_elements(con)
+    all_events = read_all_events(con)
+    elements_by_code = {c: all_elements.get(c, []) for c in codes}
+    events_by_code = {c: all_events.get(c, []) for c in codes}
+    descriptions = {
+        r[0]: r[1] or ""
+        for r in con.execute(
+            "SELECT hcpcs, arg_max(description, year) FROM pfs.rvu WHERE mod IS NULL OR mod = '' GROUP BY hcpcs"
+        ).fetchall()
+    }
+    return derive_families(elements_by_code, events_by_code, descriptions)
+
+
+def _print_families(rows: list[Any]) -> None:
+    by_key: dict[str, list[str]] = {}
+    for r in rows:
+        by_key.setdefault(r.key, []).append(f"{r.code}({r.role})")
+    for key, members in sorted(by_key.items()):
+        typer.echo(f"{key}: {' '.join(members)}")
+
+
+@app.command()
+def families(write: bool = typer.Option(False, "--write")) -> None:
+    """Derive families from pfs.code_element / pfs.code_event / pfs.rvu."""
+    if write:
+        with _batch() as con:
+            ensure_tables(con)
+            rows = _derive_rows(con)
+            _print_families(rows)
+            write_families(con, rows)
+            refresh_from(con)
+        _publish()
+        return
+    # No --write: derive-and-print never needs the write lock (I4).
+    con = _read()
+    try:
+        try:
+            rows = _derive_rows(con)
+        except Exception as exc:
+            if _missing_table(exc):
+                typer.echo("no derived tables yet — run with --write")
+                return
+            raise
+        _print_families(rows)
+    finally:
+        con.close()
+
+
+@app.command()
+def review(code: str = typer.Option("", "--code")) -> None:
+    """Element lines the classifier could not place."""
+    con = _read()
+    sql = (
+        "SELECT code, text, proposed_value, item_key, p_id FROM pfs.code_element_review"
+    )
+    params: list[Any] = []
+    if code:
+        sql += " WHERE code = ?"
+        params.append(code.upper())
+    for c, text, proposed, key, p_id in con.execute(
+        sql + " ORDER BY code, p_id", params
+    ).fetchall():
+        typer.echo(f"{c}  {key} ¶{p_id}  [{proposed or '?'}]  {text[:120]}")

src/llm/classify.py

@@ -0,0 +1,85 @@
+"""Closed-vocabulary classification on the local model pool.
+
+Given a text and a list of allowed slugs, the model must answer with
+exactly one slug or ``none``. Anything else is treated as ``None`` so a
+caller can queue the text for human review instead of inventing a value.
+Self-hosted inference only (Ollama ``/api/chat`` on the largest live
+host, the same route ``llm.rag.stream_answer`` takes).
+"""
+
+from __future__ import annotations
+
+import logging
+from typing import Any, Callable, Sequence
+
+import httpx
+
+from llm.pool import pick_model
+
+log = logging.getLogger(__name__)
+
+Classifier = Callable[[str, Sequence[str]], str | None]
+
+_SYSTEM = (
+    "You label short healthcare-policy text with exactly one slug from a "
+    "closed list. Reply with the slug only — no words, no punctuation. If no "
+    "slug fits, reply: none"
+)
+_TIMEOUT = httpx.Timeout(120.0, connect=10.0)
+
+
+def build_prompt(text: str, choices: Sequence[str]) -> list[dict]:
+    listing = "\n".join(f"- {c}" for c in choices)
+    return [
+        {"role": "system", "content": _SYSTEM},
+        {
+            "role": "user",
+            "content": f"Allowed slugs:\n{listing}\n- none\n\nText:\n{text.strip()}\n\nSlug:",
+        },
+    ]
+
+
+def parse_choice(reply: str, choices: Sequence[str]) -> str | None:
+    token = (reply or "").strip().strip("\"'`.").strip().lower()
+    if token == "none":
+        return None
+    return token if token in {c.lower() for c in choices} else None
+
+
+def _default_post(url: str, json: dict) -> dict:
+    with httpx.Client(timeout=_TIMEOUT) as client:
+        resp = client.post(url, json=json)
+        resp.raise_for_status()
+        return resp.json()
+
+
+def closed_vocab_classifier(
+    cfg: Any, pool: Any, *, post: Callable[..., dict] | None = None
+) -> Classifier:
+    """Return ``classify(text, choices) -> slug | None`` bound to *pool*.
+
+    ``pool.check`` runs once here, not per call (I5) — it's a host-liveness
+    probe, not per-request work, and a code with hundreds of element lines
+    used to re-probe the pool hundreds of times."""
+    send = post or _default_post
+    pool.check(cfg.instruct_model)
+
+    def classify(text: str, choices: Sequence[str]) -> str | None:
+        with pool.acquire_generation() as host:
+            model = pick_model(cfg, pool, host)
+            data = send(
+                f"{host}/api/chat",
+                json={
+                    "model": model,
+                    "messages": build_prompt(text, choices),
+                    "stream": False,
+                    "options": {"num_ctx": cfg.chat_num_ctx, "temperature": 0},
+                },
+            )
+        reply = data.get("message", {}).get("content", "")
+        choice = parse_choice(reply, choices)
+        if choice is None and reply.strip().lower() != "none":
+            log.info("classify: unparseable reply %r for %r", reply[:60], text[:60])
+        return choice
+
+    return classify

src/pfs/codetables.py

@@ -0,0 +1,173 @@
+"""DDL and writers for the code-family tables in the ``pfs`` schema.
+
+``pfs.code_element``        one row per (code, element) with its FR anchor
+``pfs.code_element_review`` element lines the classifier could not place
+``pfs.code_event``          dated lineage events (created, replaced, …)
+``pfs.code_family``         derived family membership with roles
+
+Writers are delete-then-insert per code (events, elements) or full
+replace (families) so a re-run is idempotent. Callers own the
+connection (``conf.connect.duckdb_batch`` for writes) and republish the
+replica afterwards; nothing here opens a database.
+"""
+
+from __future__ import annotations
+
+from dataclasses import astuple, dataclass
+from typing import Any, Sequence
+
+
+@dataclass(frozen=True)
+class ElementRow:
+    code: str
+    year: int
+    type: str
+    value: str
+    detail: str
+    text: str
+    item_key: str
+    p_id: int
+    page: int
+    source: str  # "fr" | "hcpcs" | "rvu"
+
+
+@dataclass(frozen=True)
+class ReviewRow:
+    code: str
+    text: str
+    proposed_type: str
+    proposed_value: str
+    item_key: str
+    p_id: int
+
+
+@dataclass(frozen=True)
+class EventRow:
+    code: str
+    year: int
+    kind: str
+    from_codes: str  # space-joined
+    to_codes: str
+    item_key: str
+    p_id: int
+    page: int
+    source: str  # "fr" | "rvu"
+    anchored: bool
+    note: str
+
+
+@dataclass(frozen=True)
+class FamilyRow:
+    key: str
+    name: str
+    code: str
+    role: str  # base | add-on | predecessor | successor | bundle | member
+    since: int | None
+    until: int | None
+    item_key: str
+    p_id: int
+
+
+_DDL = """
+CREATE SCHEMA IF NOT EXISTS pfs;
+CREATE TABLE IF NOT EXISTS pfs.code_element (
+    code VARCHAR, year INTEGER, type VARCHAR, value VARCHAR, detail VARCHAR,
+    text VARCHAR, item_key VARCHAR, p_id INTEGER, page INTEGER, source VARCHAR);
+CREATE TABLE IF NOT EXISTS pfs.code_element_review (
+    code VARCHAR, text VARCHAR, proposed_type VARCHAR, proposed_value VARCHAR,
+    item_key VARCHAR, p_id INTEGER);
+CREATE TABLE IF NOT EXISTS pfs.code_event (
+    code VARCHAR, year INTEGER, kind VARCHAR, from_codes VARCHAR, to_codes VARCHAR,
+    item_key VARCHAR, p_id INTEGER, page INTEGER, source VARCHAR, anchored BOOLEAN, note VARCHAR);
+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);
+"""
+
+
+def ensure_tables(con: Any) -> None:
+    for stmt in _DDL.strip().split(";"):
+        if stmt.strip():
+            con.execute(stmt)
+
+
+def _insert(con: Any, table: str, rows: Sequence[Any]) -> int:
+    if not rows:
+        return 0
+    width = len(astuple(rows[0]))
+    marks = ",".join("?" * width)
+    con.executemany(f"INSERT INTO {table} VALUES ({marks})", [astuple(r) for r in rows])
+    return len(rows)
+
+
+def write_elements(
+    con: Any, code: str, rows: Sequence[ElementRow], reviews: Sequence[ReviewRow]
+) -> int:
+    ensure_tables(con)
+    con.execute("DELETE FROM pfs.code_element WHERE code = ?", [code])
+    con.execute("DELETE FROM pfs.code_element_review WHERE code = ?", [code])
+    _insert(con, "pfs.code_element_review", reviews)
+    return _insert(con, "pfs.code_element", rows)
+
+
+def write_events(con: Any, code: str, rows: Sequence[EventRow]) -> int:
+    ensure_tables(con)
+    con.execute("DELETE FROM pfs.code_event WHERE code = ?", [code])
+    return _insert(con, "pfs.code_event", rows)
+
+
+def write_families(con: Any, rows: Sequence[FamilyRow]) -> int:
+    ensure_tables(con)
+    con.execute("DELETE FROM pfs.code_family")
+    return _insert(con, "pfs.code_family", rows)
+
+
+def read_elements(con: Any, code: str) -> list[ElementRow]:
+    rows = con.execute(
+        "SELECT * FROM pfs.code_element WHERE code = ? ORDER BY year, type, value, detail",
+        [code],
+    ).fetchall()
+    return [ElementRow(*r) for r in rows]
+
+
+def read_events(con: Any, code: str) -> list[EventRow]:
+    rows = con.execute(
+        "SELECT * FROM pfs.code_event WHERE code = ? ORDER BY year, kind", [code]
+    ).fetchall()
+    return [EventRow(*r) for r in rows]
+
+
+def read_all_elements(con: Any) -> dict[str, list[ElementRow]]:
+    """Every ``pfs.code_element`` row, grouped by code — one full-table
+    scan instead of the ~N per-code queries ``families --write`` used to
+    run inside the DuckDB write lock (I9). Rows within each code are
+    ordered exactly as ``read_elements`` orders them."""
+    rows = con.execute(
+        "SELECT * FROM pfs.code_element ORDER BY code, year, type, value, detail"
+    ).fetchall()
+    out: dict[str, list[ElementRow]] = {}
+    for r in rows:
+        row = ElementRow(*r)
+        out.setdefault(row.code, []).append(row)
+    return out
+
+
+def read_all_events(con: Any) -> dict[str, list[EventRow]]:
+    """Every ``pfs.code_event`` row, grouped by code — see
+    ``read_all_elements``. Rows within each code are ordered exactly as
+    ``read_events`` orders them."""
+    rows = con.execute(
+        "SELECT * FROM pfs.code_event ORDER BY code, year, kind"
+    ).fetchall()
+    out: dict[str, list[EventRow]] = {}
+    for r in rows:
+        row = EventRow(*r)
+        out.setdefault(row.code, []).append(row)
+    return out
+
+
+def read_families(con: Any) -> list[FamilyRow]:
+    rows = con.execute(
+        "SELECT * FROM pfs.code_family ORDER BY key, role, code"
+    ).fetchall()
+    return [FamilyRow(*r) for r in rows]

src/pfs/descriptors.py

@@ -0,0 +1,132 @@
+"""Where a code's descriptor text comes from.
+
+1. The Federal Register prints a descriptor as a *stem* paragraph
+   ("…CPT code 99490 (Chronic care management services … with the
+   following required elements:") followed by one paragraph per element
+   (``fr_anchors`` keeps them as consecutive ``p_id`` rows — CY2015 final
+   rule ¶1244–1247 for 99490, CY2025 final rule ¶1163–1185 for G0556).
+2. ``terminology.hcpcs_level_2.long_description`` for HCPCS level II
+   codes (G-codes carry the full element list; CPT long descriptors are
+   not distributed).
+3. ``pfs.rvu.description`` — the short descriptor per year (tracks
+   renames such as "Chron care mgmt srvc 20 min" → "Chrnc care mgmt
+   staff 1st 20").
+
+Function-level ``bib`` access only (a ``Store``-like object exposing
+``_con()``) so this module imports without SQLite files present.
+"""
+
+from __future__ import annotations
+
+import re
+from dataclasses import dataclass
+from typing import Any
+
+from pfs.elements import is_element_paragraph
+
+
+@dataclass(frozen=True)
+class Para:
+    item_key: str
+    p_id: int
+    page: int
+    text: str
+
+
+@dataclass(frozen=True)
+class DescriptorRun:
+    code: str
+    item_key: str
+    rule_year: int
+    stem: Para
+    elements: tuple[Para, ...]
+
+    @property
+    def text(self) -> str:
+        return " ".join([self.stem.text, *(p.text for p in self.elements)])
+
+
+_RE_CY = re.compile(r"\bCY\s?(\d{4})\b")
+
+
+def rule_year_of(title: str, date_published: str) -> int:
+    """The payment year a rule governs: ``CY nnnn`` in the title, else the
+    publication year + 1 (PFS rules publish in the second half of the
+    prior year)."""
+    m = _RE_CY.search(title or "")
+    if m:
+        return int(m.group(1))
+    return int((date_published or "1970")[:4]) + 1
+
+
+def _stem_pattern(code: str) -> re.Pattern[str]:
+    # "code 99490 (Chronic…", "CPT code 99490 (", "HCPCS code G0556 ( Advanced…"
+    # Case-insensitive on purpose (some rules print descriptors lowercase:
+    # "CPT code 99490 (chronic care management services…"), but the
+    # negative lookahead rejects the cross-reference/enumeration form
+    # "( 9) 99439 (code for non-complex chronic care management)." — that
+    # opening "(code..." is never a descriptor stem.
+    return re.compile(rf"\b{re.escape(code)}\s*\(\s*(?!codes?\b)[A-Za-z]", re.I)
+
+
+def descriptor_runs(
+    store: Any, code: str, *, max_elements: int = 40
+) -> list[DescriptorRun]:
+    """Every FR paragraph that opens *code*'s descriptor, with the element
+    paragraphs that follow it (stops at the first prose paragraph)."""
+    con = store._con()
+    pat = _stem_pattern(code.upper())
+    stems = con.execute(
+        "SELECT a.item_key, a.p_id, a.page, a.text, i.title, i.date_published "
+        "FROM fr_anchors a JOIN items i ON i.key = a.item_key "
+        "WHERE a.text LIKE ? ORDER BY i.date_published, a.p_id",
+        (f"%{code.upper()}%",),
+    ).fetchall()
+    runs: list[DescriptorRun] = []
+    for row in stems:
+        if not pat.search(row["text"]):
+            continue
+        stem = Para(row["item_key"], row["p_id"], row["page"], row["text"])
+        following = con.execute(
+            "SELECT p_id, page, text FROM fr_anchors WHERE item_key = ? AND p_id > ? "
+            "ORDER BY p_id LIMIT ?",
+            (stem.item_key, stem.p_id, max_elements),
+        ).fetchall()
+        elements: list[Para] = []
+        for f in following:
+            if not is_element_paragraph(f["text"]):
+                break
+            elements.append(Para(stem.item_key, f["p_id"], f["page"], f["text"]))
+            if f["text"].rstrip().endswith((")", ").")):
+                break  # closing parenthesis ends the descriptor
+        runs.append(
+            DescriptorRun(
+                code=code.upper(),
+                item_key=stem.item_key,
+                rule_year=rule_year_of(row["title"], row["date_published"]),
+                stem=stem,
+                elements=tuple(elements),
+            )
+        )
+    return runs
+
+
+def hcpcs_long_description(con: Any, code: str) -> str:
+    row = con.execute(
+        "SELECT long_description FROM terminology.hcpcs_level_2 WHERE hcpcs = ? ORDER BY CAST(seqnum AS INTEGER) DESC, recid DESC LIMIT 1",
+        [code.upper()],
+    ).fetchone()
+    return (row[0] or "") if row else ""
+
+
+def rvu_descriptions(con: Any, code: str) -> list[tuple[int, str, str]]:
+    """(year, status_code, description) for the base (no-modifier) row of
+    each year, ascending."""
+    rows = con.execute(
+        "SELECT year, status_code, description FROM pfs.rvu "
+        "WHERE hcpcs = ? AND (mod IS NULL OR mod = '') "
+        "QUALIFY row_number() OVER (PARTITION BY year ORDER BY mod NULLS FIRST) = 1 "
+        "ORDER BY year",
+        [code.upper()],
+    ).fetchall()
+    return [(int(y), s or "", d or "") for y, s, d in rows]

src/pfs/elements.py

@@ -0,0 +1,360 @@
+"""Closed, typed vocabulary for the logical elements of a PFS code
+descriptor, and a deterministic parser for the parts of a descriptor that
+regular expressions can read (minutes, periods, code references, the
+recurring actor / population / activity / modality phrases).
+
+CMS treats a code as a bundle of elements — telehealth Step 3 is "Review
+the elements of the service as described by the HCPCS code" (90 FR 32389).
+Types are fixed here; *values* are a curated list that grows only by
+review (an unknown phrase is queued, see ``pfs.extract``). Pure: no I/O,
+importable inside the ``llm`` container.
+"""
+
+from __future__ import annotations
+
+import re
+from dataclasses import dataclass
+from enum import Enum
+
+from pfs.families import find_codes
+
+
+class ElementType(str, Enum):
+    ACTOR = "actor"
+    TIME = "time"
+    PERIOD = "period"
+    POPULATION = "population"
+    ACTIVITY = "activity"
+    MODALITY = "modality"
+    RELATION = "relation"
+    SETTING = "setting"
+    BILLING = "billing"
+
+
+@dataclass(frozen=True, order=True)
+class Element:
+    type: ElementType
+    value: str
+    detail: str = ""  # "20" for TIME, the target code for RELATION, "" otherwise
+
+
+#: Closed value lists. Seeded from the descriptors the FR prints for
+#: 99490/99487/99489/99491/99437/99439, 99495/99496, 99497/99498,
+#: 99424–99427, G0556–G0558, 99441–99443/98008–98016, G2211, 99457/99458.
+VOCAB: dict[ElementType, tuple[str, ...]] = {
+    ElementType.ACTOR: (
+        "clinical-staff-directed",
+        "physician-or-qhp-personally",
+        "rhc-fqhc",
+        "care-team",
+    ),
+    ElementType.TIME: ("first", "each-additional", "at-least", "total"),
+    ElementType.PERIOD: (
+        "calendar-month",
+        "30-days",
+        "per-visit",
+        "14-days-post-discharge",
+        "7-days-post-discharge",
+        "per-day",
+    ),
+    ElementType.POPULATION: (
+        "multiple-chronic-conditions-12-months",
+        "single-high-risk-condition",
+        "significant-risk",
+        "qualified-medicare-beneficiary",
+        "new-patient",
+        "established-patient",
+        "one-or-fewer-chronic-conditions",
+    ),
+    ElementType.ACTIVITY: (
+        "comprehensive-care-plan",
+        "consent",
+        "initiating-visit",
+        "24-7-access",
+        "continuity-of-care",
+        "alternative-care-delivery",
+        "comprehensive-care-management",
+        "needs-assessment",
+        "preventive-services",
+        "medication-reconciliation",
+        "care-transitions",
+        "community-coordination",
+        "asynchronous-communication",
+        "population-data-analysis",
+        "risk-stratification",
+        "performance-measurement",
+        "medical-decision-making",
+        "interactive-contact-post-discharge",
+        "advance-directive-discussion",
+        "device-data-review",
+    ),
+    ElementType.MODALITY: (
+        "face-to-face",
+        "interactive-telecommunications",
+        "audio-video",
+        "audio-only",
+        "asynchronous",
+        "non-face-to-face",
+    ),
+    ElementType.RELATION: (
+        "addon-of",
+        "not-with",
+        "replaces",
+        "replaced-by",
+        "defined-by-reference-to",
+        "crosswalk-valued-to",
+    ),
+    ElementType.SETTING: ("non-facility", "facility", "home", "office"),
+    ElementType.BILLING: (
+        "list-separately",
+        "one-practitioner-per-month",
+        "focal-point-all-primary-care",
+        "once-per-period",
+    ),
+}
+
+_MIN = r"(\d{1,3})\s*(?:minutes|min)"
+_RE_FIRST = re.compile(rf"\b(?:first|initial)\s+{_MIN}", re.I)
+_RE_AT_LEAST = re.compile(rf"\bat least\s+{_MIN}", re.I)
+_RE_EACH_ADDL = re.compile(rf"\beach additional\s+{_MIN}", re.I)
+_RE_TOTAL = re.compile(rf"\btotal (?:time )?(?:of )?{_MIN}", re.I)
+
+_PERIOD_PATTERNS: tuple[tuple[re.Pattern[str], str], ...] = (
+    (re.compile(r"\bper calendar month\b", re.I), "calendar-month"),
+    (re.compile(r"\bper 30 days\b", re.I), "30-days"),
+    (
+        re.compile(r"\bwithin 14 (?:calendar )?days of discharge\b", re.I),
+        "14-days-post-discharge",
+    ),
+    (
+        re.compile(r"\bwithin 7 (?:calendar )?days of discharge\b", re.I),
+        "7-days-post-discharge",
+    ),
+    (re.compile(r"\bper day\b", re.I), "per-day"),
+    (re.compile(r"\bper visit\b", re.I), "per-visit"),
+)
+
+_PHRASES: tuple[tuple[ElementType, str, re.Pattern[str]], ...] = (
+    (
+        ElementType.ACTOR,
+        "clinical-staff-directed",
+        re.compile(r"clinical staff(?: time)?[^.;]{0,40}directed by a physician", re.I),
+    ),
+    (
+        ElementType.ACTOR,
+        "physician-or-qhp-personally",
+        re.compile(r"(?:provided|furnished) personally by a physician", re.I),
+    ),
+    (ElementType.ACTOR, "rhc-fqhc", re.compile(r"\b(?:RHC|FQHC)s?\b")),
+    (
+        ElementType.POPULATION,
+        "multiple-chronic-conditions-12-months",
+        re.compile(r"multiple \(two or more\) chronic conditions", re.I),
+    ),
+    (
+        ElementType.POPULATION,
+        "one-or-fewer-chronic-conditions",
+        re.compile(r"one chronic condition[^;]{0,120}or fewer", re.I),
+    ),
+    (
+        ElementType.POPULATION,
+        "significant-risk",
+        re.compile(r"significant risk of death", re.I),
+    ),
+    (
+        ElementType.POPULATION,
+        "qualified-medicare-beneficiary",
+        re.compile(r"qualified medicare beneficiary", re.I),
+    ),
+    (ElementType.POPULATION, "new-patient", re.compile(r"\bnew patient\b", re.I)),
+    (
+        ElementType.POPULATION,
+        "established-patient",
+        re.compile(r"\bestablished patient\b", re.I),
+    ),
+    (
+        ElementType.ACTIVITY,
+        "comprehensive-care-plan",
+        re.compile(r"comprehensive care plan", re.I),
+    ),
+    (ElementType.ACTIVITY, "consent", re.compile(r"\bconsent\b", re.I)),
+    (
+        ElementType.ACTIVITY,
+        "initiating-visit",
+        re.compile(r"initiat\w+ (?:during a )?(?:qualifying )?visit", re.I),
+    ),
+    (ElementType.ACTIVITY, "24-7-access", re.compile(r"24/7 access", re.I)),
+    (
+        ElementType.ACTIVITY,
+        "continuity-of-care",
+        re.compile(r"continuity of care", re.I),
+    ),
+    (
+        ElementType.ACTIVITY,
+        "alternative-care-delivery",
+        re.compile(r"alternative ways to traditional office visits", re.I),
+    ),
+    (
+        ElementType.ACTIVITY,
+        "comprehensive-care-management",
+        re.compile(r"comprehensive care management", re.I),
+    ),
+    (ElementType.ACTIVITY, "needs-assessment", re.compile(r"needs assessment", re.I)),
+    (
+        ElementType.ACTIVITY,
+        "preventive-services",
+        re.compile(r"preventive services", re.I),
+    ),
+    (
+        ElementType.ACTIVITY,
+        "medication-reconciliation",
+        re.compile(r"medication reconciliation", re.I),
+    ),
+    (
+        ElementType.ACTIVITY,
+        "care-transitions",
+        re.compile(r"care transitions|transitions? of care", re.I),
+    ),
+    (
+        ElementType.ACTIVITY,
+        "community-coordination",
+        re.compile(r"community-based", re.I),
+    ),
+    (
+        ElementType.ACTIVITY,
+        "asynchronous-communication",
+        re.compile(r"asynchronous", re.I),
+    ),
+    (
+        ElementType.ACTIVITY,
+        "population-data-analysis",
+        re.compile(r"patient population data", re.I),
+    ),
+    (ElementType.ACTIVITY, "risk-stratification", re.compile(r"risk stratif", re.I)),
+    (
+        ElementType.ACTIVITY,
+        "performance-measurement",
+        re.compile(r"performance measurement", re.I),
+    ),
+    (
+        ElementType.ACTIVITY,
+        "medical-decision-making",
+        re.compile(r"medical decision making", re.I),
+    ),
+    (
+        ElementType.ACTIVITY,
+        "interactive-contact-post-discharge",
+        re.compile(r"interactive contact", re.I),
+    ),
+    (
+        ElementType.ACTIVITY,
+        "advance-directive-discussion",
+        re.compile(r"advance directive", re.I),
+    ),
+    (
+        ElementType.ACTIVITY,
+        "device-data-review",
+        re.compile(r"physiologic (?:parameter|data)", re.I),
+    ),
+    (
+        ElementType.MODALITY,
+        "interactive-telecommunications",
+        re.compile(r"interactive telecommunications", re.I),
+    ),
+    (
+        ElementType.MODALITY,
+        "audio-video",
+        re.compile(r"audio-video|audio and video", re.I),
+    ),
+    (ElementType.MODALITY, "audio-only", re.compile(r"audio-only|telephone", re.I)),
+    (ElementType.MODALITY, "face-to-face", re.compile(r"(?<!non-)face-to-face", re.I)),
+    (ElementType.MODALITY, "non-face-to-face", re.compile(r"non-face-to-face", re.I)),
+    (ElementType.SETTING, "home", re.compile(r"\bhome visits?\b", re.I)),
+    (
+        ElementType.BILLING,
+        "list-separately",
+        re.compile(r"list separately in addition", re.I),
+    ),
+    (
+        ElementType.BILLING,
+        "one-practitioner-per-month",
+        re.compile(r"only one practitioner can furnish", re.I),
+    ),
+    (
+        ElementType.BILLING,
+        "focal-point-all-primary-care",
+        re.compile(r"continuing focal point", re.I),
+    ),
+)
+
+_RE_IN_CONJUNCTION = re.compile(r"in conjunction with\s+([^)]+)", re.I)
+_RE_NOT_WITH = re.compile(
+    r"do not report[^)]*?(?:same (?:calendar )?month|in conjunction|with)\s+(?:as\s+)?([^)]+)",
+    re.I,
+)
+_RE_ELEMENTS_OF = re.compile(
+    r"with the elements included in\s+([A-Z]\d{4}|\d{5})", re.I
+)
+
+
+def parse_descriptor(text: str) -> tuple[Element, ...]:
+    """Every element a regex can read from *text*, sorted and unique."""
+    found: set[Element] = set()
+    for pat, value in (
+        (_RE_FIRST, "first"),
+        (_RE_EACH_ADDL, "each-additional"),
+        (_RE_TOTAL, "total"),
+    ):
+        for m in pat.finditer(text):
+            found.add(Element(ElementType.TIME, value, m.group(1)))
+    for m in _RE_AT_LEAST.finditer(text):
+        # "at least N minutes" is the first-unit threshold unless an
+        # each-additional phrase already claims N.
+        value = "first" if not _RE_EACH_ADDL.search(text) else "at-least"
+        found.add(Element(ElementType.TIME, value, m.group(1)))
+    for pat, value in _PERIOD_PATTERNS:
+        if pat.search(text):
+            found.add(Element(ElementType.PERIOD, value))
+    for type_, value, pat in _PHRASES:
+        if pat.search(text):
+            found.add(Element(type_, value))
+    for m in _RE_IN_CONJUNCTION.finditer(text):
+        for code in find_codes(m.group(1)):
+            found.add(Element(ElementType.RELATION, "addon-of", code))
+    for m in _RE_NOT_WITH.finditer(text):
+        for code in find_codes(m.group(1)):
+            found.add(Element(ElementType.RELATION, "not-with", code))
+    for m in _RE_ELEMENTS_OF.finditer(text):
+        found.add(
+            Element(ElementType.RELATION, "defined-by-reference-to", m.group(1).upper())
+        )
+    return tuple(sorted(found, key=lambda e: (e.type.value, e.value, e.detail)))
+
+
+_RE_PROSE_START = re.compile(
+    r"^(?:Comment|Response|We|In the|After|For|As|The|CMS|Section)\b"
+)
+# An enumeration list item ("( 9) 99439 (code for …)") or a cross-reference
+# sentence ("… codes for remote physiologic monitoring …") is prose about a
+# code, not a fragment of that code's own descriptor — even though it can
+# end in ");" and otherwise look like a real element line (C1).
+_RE_ENUM_OR_CODE_FOR = re.compile(r"^\(\s*\d+\)|\b(?:code|codes)\s+for\b", re.I)
+
+
+def is_element_paragraph(text: str) -> bool:
+    """True for the short, fragment-like lines the FR prints inside a
+    descriptor (``Consent;``, ``++ Document …``), false for prose."""
+    t = text.strip()
+    if not t or len(t) > 600:
+        return False
+    if t.startswith("++"):
+        return True
+    if _RE_PROSE_START.match(t) or _RE_ENUM_OR_CODE_FOR.search(t):
+        return False
+    return t.endswith((";", ")", ").", ":")) or (
+        t[0].islower() is False and t.count(". ") <= 1 and len(t) < 300
+    )
+
+
+def slug(e: Element) -> str:
+    return f"{e.type.value}={e.value}" + (f":{e.detail}" if e.detail else "")

src/pfs/extract.py

@@ -0,0 +1,114 @@
+"""Element extraction for one code: deterministic parse first, the local
+model second (closed vocabulary only), a review queue for the rest.
+
+``extract_run`` works on an FR descriptor run (stem + element
+paragraphs); ``extract_text`` on a flat descriptor (HCPCS long
+description, RVU short description); ``extract_code`` gathers every
+source for a code. No I/O here beyond what the caller hands in.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Any, Callable, Sequence
+
+from pfs.codetables import ElementRow, ReviewRow
+from pfs.descriptors import (
+    DescriptorRun,
+    Para,
+    descriptor_runs,
+    hcpcs_long_description,
+    rvu_descriptions,
+)
+from pfs.elements import VOCAB, Element, ElementType, parse_descriptor
+
+Classifier = Callable[[str, Sequence[str]], "str | None"]
+
+#: Types the classifier may assign to a free-standing element line.
+_CLASSIFIABLE = (ElementType.ACTIVITY, ElementType.POPULATION, ElementType.BILLING)
+_CHOICES: tuple[str, ...] = tuple(v for t in _CLASSIFIABLE for v in VOCAB[t])
+_TYPE_OF: dict[str, ElementType] = {v: t for t in _CLASSIFIABLE for v in VOCAB[t]}
+
+
+@dataclass(frozen=True)
+class Extraction:
+    code: str
+    rows: tuple[ElementRow, ...]
+    reviews: tuple[ReviewRow, ...]
+
+
+def _row(code: str, e: Element, para: Para, *, year: int, source: str) -> ElementRow:
+    return ElementRow(
+        code,
+        year,
+        e.type.value,
+        e.value,
+        e.detail,
+        para.text,
+        para.item_key,
+        para.p_id,
+        para.page,
+        source,
+    )
+
+
+def extract_run(
+    run: DescriptorRun, *, classify: Classifier | None = None
+) -> Extraction:
+    rows: dict[Element, ElementRow] = {}
+    reviews: list[ReviewRow] = []
+    # 1. deterministic, per element paragraph (anchors to that paragraph)
+    for para in run.elements:
+        found = parse_descriptor(para.text)
+        for e in found:
+            rows.setdefault(e, _row(run.code, e, para, year=run.rule_year, source="fr"))
+        if found:
+            continue
+        # 2. classifier for lines nothing matched
+        choice = classify(para.text, _CHOICES) if classify else None
+        if choice is not None and choice in _TYPE_OF:
+            e = Element(_TYPE_OF[choice], choice)
+            rows.setdefault(e, _row(run.code, e, para, year=run.rule_year, source="fr"))
+        else:
+            reviews.append(
+                ReviewRow(run.code, para.text, "", "", para.item_key, para.p_id)
+            )
+    # 3. deterministic on the stem (and anything the whole text reveals)
+    for e in parse_descriptor(run.text):
+        rows.setdefault(e, _row(run.code, e, run.stem, year=run.rule_year, source="fr"))
+    ordered = sorted(rows.values(), key=lambda r: (r.type, r.value, r.detail))
+    return Extraction(run.code, tuple(ordered), tuple(reviews))
+
+
+def extract_text(code: str, text: str, *, year: int, source: str) -> Extraction:
+    rows = [
+        ElementRow(code, year, e.type.value, e.value, e.detail, text, "", 0, 0, source)
+        for e in parse_descriptor(text)
+    ]
+    return Extraction(code, tuple(rows), ())
+
+
+def extract_code(
+    store: Any, con: Any, code: str, *, classify: Classifier | None = None
+) -> Extraction:
+    """All sources for *code*; FR rows win over HCPCS/RVU rows for the same
+    element, and each source contributes its own year."""
+    code = code.upper()
+    merged: dict[tuple[str, str, str], ElementRow] = {}
+    reviews: list[ReviewRow] = []
+    for run in descriptor_runs(store, code):
+        x = extract_run(run, classify=classify)
+        for r in x.rows:
+            merged.setdefault((r.type, r.value, r.detail), r)
+        reviews.extend(x.reviews)
+    long_desc = hcpcs_long_description(con, code)
+    years = rvu_descriptions(con, code)
+    if long_desc:
+        y = years[-1][0] if years else 0
+        for r in extract_text(code, long_desc, year=y, source="hcpcs").rows:
+            merged.setdefault((r.type, r.value, r.detail), r)
+    for year, _status, desc in years:
+        for r in extract_text(code, desc, year=year, source="rvu").rows:
+            merged.setdefault((r.type, r.value, r.detail), r)
+    ordered = sorted(merged.values(), key=lambda r: (r.type, r.value, r.detail))
+    return Extraction(code, tuple(ordered), tuple(reviews))

src/pfs/families.py

@@ -8,8 +8,12 @@ DuckDB, no narwhals — it must import inside the ``llm`` container.
 
 from __future__ import annotations
 
+import logging
 import re
 from dataclasses import dataclass
+from typing import Any, Mapping, Sequence
+
+_logger = logging.getLogger(__name__)
 
 # HCPCS level II (letter + 4 digits) or CPT (5 digits). Validation against
 # the fee schedule happens at lookup time (pfs.valuation), not here.
@@ -18,6 +22,9 @@ CODE_RE = re.compile(r"\b(?:[A-Z]\d{4}|\d{5})\b", re.IGNORECASE)
 # "91 FR 43842" — an FR page number is five digits and so reads as a CPT
 # code; citations are removed before scanning so they never become one.
 _FR_CITE_RE = re.compile(r"\b\d{1,3}\s+FR\s+\d{3,6}\b", re.IGNORECASE)
+#: Public alias — ``pfs.lineage.fr_events`` reuses this so a citation's
+#: page number never triggers an event pattern (e.g. "crosswalk") either.
+FR_CITE_RE = _FR_CITE_RE
 
 
 def find_codes(text: str) -> tuple[str, ...]:
@@ -39,7 +46,7 @@ class Family:
     synonyms: tuple[str, ...]  # lower-case; matched on word boundaries
 
 
-FAMILIES: dict[str, Family] = {
+HAND_FAMILIES: dict[str, Family] = {
     "ACP": Family(
         "ACP",
         "Advance Care Planning",
@@ -72,11 +79,25 @@ FAMILIES: dict[str, Family] = {
     ),
 }
 
-_CODE_TO_FAMILY: dict[str, Family] = {c: f for f in FAMILIES.values() for c in f.codes}
+#: Live registry. Starts as the hand list; ``refresh_from(con)`` merges in
+#: the derived families (``pfs.code_family``) in place — hand keys seeded
+#: first, derived rows extend/override per key — so every importer sees
+#: the same dict object. Only the ``stack pfs`` CLI (``elements``,
+#: ``lineage``, ``families``) calls ``refresh_from`` today; the spec's
+#: §Components asked for an import-time load, but that needs a live
+#: DuckDB connection, which is forbidden at import (the ``llm`` chat
+#: container must import this module with no database reachable). Until
+#: something calls ``refresh_from`` in that path too — a controller
+#: follow-up issue — the chat sees hand families only.
+FAMILIES: dict[str, Family] = dict(HAND_FAMILIES)
 
 
 def family_of(code: str) -> Family | None:
-    return _CODE_TO_FAMILY.get(code.upper())
+    code = code.upper()
+    for fam in FAMILIES.values():
+        if code in fam.codes:
+            return fam
+    return None
 
 
 @dataclass(frozen=True)
@@ -107,3 +128,347 @@ def detect_codes(text: str) -> Detection:
         families=tuple(sorted(families)),
         explicit=explicit,
     )
+
+
+_NOISE = frozenset(
+    {
+        "1st",
+        "ea",
+        "addl",
+        "add",
+        "min",
+        "mo",
+        "staff",
+        "staf",
+        "phys",
+        "qhp",
+        "lvl",
+        "svc",
+        "srvc",
+        "svcs",
+        "w",
+        "w/o",
+        "pt",
+        "per",
+        "the",
+        "a",
+        "of",
+        "&",
+        "each",
+        "first",
+        "initial",
+        "additional",
+        "level",
+    }
+)
+_RE_TOKEN = re.compile(r"(?<![a-z0-9])[a-z]+(?:/[a-z]+)?(?![a-z0-9])")
+
+
+def stem_tokens(description: str | None) -> frozenset[str]:
+    """Short-descriptor tokens minus time/actor/number noise — the part
+    of a description that names the service.
+
+    ``description`` may be ``None`` — a real ``pfs.rvu`` row can carry a
+    NULL description (a corpus parsing artifact, not a missing key), and
+    the caller's ``dict.get(c, default)`` only applies its default when
+    ``c`` is absent, not when it maps to ``None``."""
+    return frozenset(
+        t
+        for t in _RE_TOKEN.findall((description or "").lower())
+        if t not in _NOISE and len(t) > 1
+    )
+
+
+# Deliberately narrow (#687 over-merge incident): a family-membership edge
+# is only ever a direct, unambiguous statement that two codes belong
+# together. Dropped from the relation set: ``not-with`` (an explicit
+# *exclusion*, the opposite of membership), ``replaces``/``replaced-by``/
+# ``crosswalk-valued-to`` relation *elements* (these overlap the lineage
+# *event* signal below, which already carries year/anchor precision).
+# ``crosswalk`` events are dropped entirely — a rule's "codes ... are
+# crosswalked to X for valuation" sentence routinely lists a whole
+# family's worth of codes together and is evidence, not a merge signal.
+_LINK_RELATIONS = {"addon-of", "defined-by-reference-to"}
+#: ``replaces``/``replaced_by`` events still count, but only when the
+#: relevant field names exactly one code — a multi-code list (a blanket
+#: "these codes are replaced by ..." sentence spanning several families,
+#: see #687) is ambiguous and is ignored as a merge edge.
+_LINK_EVENTS = {"replaces", "replaced_by"}
+
+
+def _jaccard(a: frozenset[str], b: frozenset[str]) -> float:
+    return len(a & b) / len(a | b) if a or b else 0.0
+
+
+def _name_group(
+    members: Sequence[str],
+    stems: Mapping[str, frozenset[str]],
+    descriptions: Mapping[str, str],
+    tenure: Any,
+) -> tuple[str, str, str]:
+    """The stem-derived ``(key, name, representative code)`` for a group
+    with no hand family — shared by ungrouped components and by the
+    leftover codes a hand-family split (see ``derive_families``) doesn't
+    reach. The representative code is returned so a caller can disambiguate
+    a key collision (#687 live corpus: 627 stem keys shared by >= 2
+    unrelated components, e.g. ``GENE`` x13)."""
+    rep = sorted(members, key=lambda c: (-tenure(c), c))[0]
+    key = (
+        "-".join(
+            t
+            for t in _RE_TOKEN.findall((descriptions.get(rep) or "").lower())
+            if t in stems[rep]
+        ).upper()
+        or rep
+    )
+    name = descriptions.get(rep) or rep
+    return key, name, rep
+
+
+def _unique_key(key: str, rep: str, used_keys: set[str]) -> str:
+    """*key*, or ``key-rep`` if *key* is already taken by another group —
+    every non-hand key ``derive_families`` hands out must be unique."""
+    if key in used_keys:
+        key = f"{key}-{rep}"
+    used_keys.add(key)
+    return key
+
+
+def _group_rows(
+    key: str,
+    name: str,
+    members: Sequence[str],
+    elements: Mapping[str, Sequence[Any]],
+    events: Mapping[str, Sequence[Any]],
+) -> list[Any]:
+    """``FamilyRow`` per code in *members*, labeled ``key``/``name``."""
+    from pfs.codetables import FamilyRow
+
+    out = []
+    for c in sorted(members):
+        els = elements.get(c, ())
+        evs = events.get(c, ())
+        kinds = {ev.kind for ev in evs}
+        if "replaced_by" in kinds:
+            role = "predecessor"
+        elif any(e.type == "relation" and e.value == "addon-of" for e in els):
+            role = "add-on"
+        elif "replaces" in kinds:
+            role = "successor"
+        else:
+            role = "base"
+        since = next((ev.year for ev in evs if ev.kind == "appeared"), None)
+        until = next((ev.year for ev in evs if ev.kind == "disappeared"), None)
+        anchor = next(((ev.item_key, ev.p_id) for ev in evs if ev.item_key), ("", 0))
+        out.append(FamilyRow(key, name, c, role, since, until, anchor[0], anchor[1]))
+    return out
+
+
+def _connected(nodes: Sequence[str], adj: Mapping[str, set[str]]) -> list[list[str]]:
+    """Connected components of *nodes* using only edges between members of
+    *nodes* itself (an edge to a node outside the set doesn't count)."""
+    node_set = set(nodes)
+    seen: set[str] = set()
+    out: list[list[str]] = []
+    for start in sorted(nodes):
+        if start in seen:
+            continue
+        stack = [start]
+        seen.add(start)
+        comp = []
+        while stack:
+            c = stack.pop()
+            comp.append(c)
+            for n in adj.get(c, ()):
+                if n in node_set and n not in seen:
+                    seen.add(n)
+                    stack.append(n)
+        out.append(comp)
+    return out
+
+
+def derive_families(
+    elements: Mapping[str, Sequence[Any]],
+    events: Mapping[str, Sequence[Any]],
+    descriptions: Mapping[str, str],
+) -> list[Any]:
+    """Connected components over codes linked by relation elements,
+    lineage events, or a shared service stem + activity set.
+
+    Role precedence when a code qualifies for more than one:
+    predecessor (has a ``replaced_by`` event) > add-on (has an
+    ``addon-of`` relation element) > successor (has a ``replaces``
+    event) > base.
+
+    A component that reaches into two or more hand families (#687: the
+    real corpus's blanket multi-code lineage sentences used to merge ACP,
+    CCM and PCM into one blob even with the narrowed edges above still
+    occasionally bridging two families) is split: each hand family's own
+    codes seed a BFS over the component's edges, run in ``HAND_FAMILIES``
+    order, and every code the BFS reaches joins that hand family. Codes no
+    BFS reaches keep their own stem-keyed family, same as an ungrouped
+    component.
+    """
+    # Reject members that aren't code-shaped before deriving anything —
+    # a corpus parsing artifact (e.g. a bare '\x1a') must never become a
+    # family of its own.
+    codes = sorted(
+        c
+        for c in (set(elements) | set(events) | set(descriptions))
+        if CODE_RE.fullmatch(c)
+    )
+    parent = {c: c for c in codes}
+    adj: dict[str, set[str]] = {c: set() for c in codes}
+
+    def find(c: str) -> str:
+        while parent[c] != c:
+            parent[c] = parent[parent[c]]
+            c = parent[c]
+        return c
+
+    def union(a: str, b: str) -> None:
+        if a in parent and b in parent:
+            adj[a].add(b)
+            adj[b].add(a)
+            parent[find(a)] = find(b)
+
+    for code, els in elements.items():
+        for e in els:
+            if e.type == "relation" and e.value in _LINK_RELATIONS and e.detail:
+                union(code, e.detail.upper())
+    for code, evs in events.items():
+        for ev in evs:
+            if ev.kind not in _LINK_EVENTS:
+                continue
+            field = ev.from_codes if ev.kind == "replaces" else ev.to_codes
+            targets = field.split()
+            if len(targets) == 1:
+                union(code, targets[0].upper())
+    stems = {c: stem_tokens(descriptions.get(c) or "") for c in codes}
+    acts = {
+        c: frozenset(e.value for e in elements.get(c, ()) if e.type == "activity")
+        for c in codes
+    }
+    # Inverted index token -> codes, so we only compare pairs that share at
+    # least one stem token. Exact for the >= 0.5 threshold: any qualifying
+    # pair has a non-empty intersection (jaccard > 0 implies shared tokens),
+    # so no comparison is skipped, but the O(n^2) all-pairs scan is.
+    buckets: dict[str, list[str]] = {}
+    for c, toks in stems.items():
+        for t in toks:
+            buckets.setdefault(t, []).append(c)
+    seen: set[tuple[str, str]] = set()
+    for members in buckets.values():
+        for i, a in enumerate(members):
+            for b in members[i + 1 :]:
+                pair = (a, b) if a < b else (b, a)
+                if pair in seen:
+                    continue
+                seen.add(pair)
+                if (
+                    acts[a]
+                    and acts[a] == acts[b]
+                    and _jaccard(stems[a], stems[b]) >= 0.5
+                ):
+                    union(a, b)
+
+    groups: dict[str, list[str]] = {}
+    for c in codes:
+        groups.setdefault(find(c), []).append(c)
+
+    def tenure(c: str) -> int:
+        since = next((e.year for e in events.get(c, ()) if e.kind == "appeared"), 0)
+        until = next(
+            (e.year for e in events.get(c, ()) if e.kind == "disappeared"), 9999
+        )
+        return until - since
+
+    rows: list[Any] = []
+    used_keys: set[str] = set()
+    for members in groups.values():
+        member_set = set(members)
+        hands = [k for k in HAND_FAMILIES if set(HAND_FAMILIES[k].codes) & member_set]
+        if len(hands) >= 2:
+            _logger.warning(
+                "family component of %d codes spans hand families %s — "
+                "splitting by hand-family BFS",
+                len(members),
+                hands,
+            )
+            assigned: dict[str, str] = {}
+            for hk in hands:
+                for seed in HAND_FAMILIES[hk].codes:
+                    if seed in member_set:
+                        assigned[seed] = hk
+            for hk in hands:
+                frontier = [c for c in HAND_FAMILIES[hk].codes if c in member_set]
+                while frontier:
+                    nxt = []
+                    for c in frontier:
+                        for n in adj.get(c, ()):
+                            if n in member_set and n not in assigned:
+                                assigned[n] = hk
+                                nxt.append(n)
+                    frontier = nxt
+            for hk in hands:
+                grp = [c for c, v in assigned.items() if v == hk]
+                if grp:
+                    rows.extend(
+                        _group_rows(hk, HAND_FAMILIES[hk].name, grp, elements, events)
+                    )
+            unreached = [c for c in members if c not in assigned]
+            for sub in _connected(unreached, adj):
+                key, name, rep = _name_group(sub, stems, descriptions, tenure)
+                key = _unique_key(key, rep, used_keys)
+                rows.extend(_group_rows(key, name, sub, elements, events))
+            continue
+        hand = hands[0] if hands else None
+        if hand:
+            key, name = hand, HAND_FAMILIES[hand].name
+        else:
+            key, name, rep = _name_group(members, stems, descriptions, tenure)
+            key = _unique_key(key, rep, used_keys)
+        rows.extend(_group_rows(key, name, members, elements, events))
+    return sorted(rows, key=lambda r: (r.key, r.code))
+
+
+def load_families(con: Any) -> dict[str, Family]:
+    from pfs.codetables import read_families
+
+    try:
+        rows = read_families(con)
+    except Exception as exc:  # table/schema missing on an old replica
+        msg = str(exc)
+        if "code_family" in msg and ("Catalog" in msg or "does not exist" in msg):
+            _logger.info("pfs.code_family not present; no derived families: %s", exc)
+            return {}
+        raise
+    out: dict[str, Family] = {}
+    for r in rows:
+        fam = out.get(r.key)
+        codes = (*(fam.codes if fam else ()), r.code)
+        syn = (
+            HAND_FAMILIES[r.key].synonyms
+            if r.key in HAND_FAMILIES
+            else (r.key.lower(), r.name.lower())
+        )
+        out[r.key] = Family(r.key, r.name, tuple(sorted(set(codes))), syn)
+    return out
+
+
+def refresh_from(con: Any) -> int:
+    """Merge ``pfs.code_family`` into the live registry: hand families
+    seed it first, so a hand key the derivation table doesn't reach still
+    survives; a hand key the table *does* reach is replaced by the
+    derived (superset) codes — ``load_families`` already carries the
+    hand's own name/synonyms forward for that key (Ruling 17: a bare
+    ``FAMILIES.clear()`` used to wipe every hand family the moment any
+    derived row loaded)."""
+    derived = load_families(con)
+    if not derived:
+        return 0
+    merged = dict(HAND_FAMILIES)
+    merged.update(derived)
+    FAMILIES.clear()
+    FAMILIES.update(merged)
+    return len(FAMILIES)

src/pfs/lineage.py

@@ -0,0 +1,267 @@
+"""Lineage events for a code: what happened, when, and where the record
+says so.
+
+Two independent sources: the RVU files (``pfs.rvu`` base rows year over
+year — appeared / disappeared / status_change / descriptor_change /
+revalued) and the Federal Register (``fr_anchors`` paragraphs that name
+the code next to an event verb — created / adopted_cpt / replaces /
+replaced_by / deleted / crosswalk / bundled / telehealth_list).
+``lineage`` merges them and marks each RVU event ``anchored`` when an FR
+event for the same code lies within ±1 rule year; the rest are
+``unanchored`` for review, never silently accepted.
+"""
+
+from __future__ import annotations
+
+import dataclasses
+import re
+from typing import Any
+
+from pfs.codetables import EventRow
+from pfs.descriptors import rule_year_of
+from pfs.families import FR_CITE_RE, find_codes
+
+RVU_KINDS = (
+    "appeared",
+    "disappeared",
+    "status_change",
+    "descriptor_change",
+    "revalued",
+)
+
+
+def rvu_year_span(con: Any) -> tuple[int, int]:
+    lo, hi = con.execute("SELECT min(year), max(year) FROM pfs.rvu").fetchone()
+    return int(lo), int(hi)
+
+
+def _base_rows(con: Any, code: str) -> list[tuple[int, str, str, float | None]]:
+    return [
+        (int(y), s or "", d or "", t)
+        for y, s, d, t in con.execute(
+            "SELECT year, status_code, description, non_fac_total FROM pfs.rvu "
+            "WHERE hcpcs = ? AND (mod IS NULL OR mod = '') "
+            "QUALIFY row_number() OVER (PARTITION BY year ORDER BY mod NULLS FIRST) = 1 "
+            "ORDER BY year",
+            [code.upper()],
+        ).fetchall()
+    ]
+
+
+def _ev(code: str, year: int, kind: str, note: str = "") -> EventRow:
+    return EventRow(code.upper(), year, kind, "", "", "", 0, 0, "rvu", False, note)
+
+
+def rvu_events(
+    con: Any, code: str, *, revalue_threshold: float = 0.10
+) -> list[EventRow]:
+    rows = _base_rows(con, code)
+    if not rows:
+        return []
+    lo, hi = rvu_year_span(con)
+    out: list[EventRow] = []
+    first, last = rows[0][0], rows[-1][0]
+    if first > lo:
+        out.append(_ev(code, first, "appeared"))
+    if last < hi:
+        out.append(_ev(code, last + 1, "disappeared"))
+    for (py, ps, pd, pt), (y, s, d, t) in zip(rows, rows[1:]):
+        if s != ps:
+            out.append(_ev(code, y, "status_change", f"{ps}→{s}"))
+        if d != pd:
+            out.append(_ev(code, y, "descriptor_change", f"{pd} → {d}"))
+        if pt and t is not None and pt > 0:
+            delta = (t - pt) / pt
+            if abs(delta) > revalue_threshold:
+                out.append(_ev(code, y, "revalued", f"{delta:+.1%}"))
+    return sorted(out, key=lambda e: (e.year, RVU_KINDS.index(e.kind)))
+
+
+FR_KINDS = (
+    "created",
+    "adopted_cpt",
+    "replaces",
+    "replaced_by",
+    "deleted",
+    "crosswalk",
+    "bundled",
+    "telehealth_list",
+)
+_CODE = r"(?:[A-Z]\d{4}|\d{5})"
+
+
+def _patterns(code: str) -> list[tuple[str, re.Pattern[str], str]]:
+    c = re.escape(code)
+    return [
+        (
+            "created",
+            re.compile(
+                rf"(?:\bcreat\w*\b.{{0,80}}\b(?:HCPCS|CPT|G-?)\s*codes?\b[^.]{{0,80}}{c})"
+                rf"|(?:new (?:HCPCS|CPT) code\s+{c})",
+                re.I,
+            ),
+            "",
+        ),
+        (
+            "adopted_cpt",
+            re.compile(rf"adopt(?:ed|ing)?\s+CPT code\s+{c}", re.I),
+            "",
+        ),
+        (
+            "replaces",
+            re.compile(
+                rf"{c}(?:(?!{_CODE})[^.]){{0,120}}"
+                rf"\b(?:replace|replacing|replacement for|instead of|with a descriptor identical to)\b",
+                re.I,
+            ),
+            "from",
+        ),
+        (
+            "replaced_by",
+            re.compile(
+                rf"(?:\b(?:replace\w*|replacement)\b(?:(?!{_CODE})[^.]){{0,120}}{c})"
+                rf"|(?:{c}[^.]{{0,80}}\b(?:as|is being|will be|which we are) (?:finaliz|replac)\w+ (?:in this rule )?as new (?:CPT|HCPCS) code\s+{_CODE})",
+                re.I,
+            ),
+            "to",
+        ),
+        (
+            "deleted",
+            re.compile(
+                rf"(?:\bdelet(?:ed|ion|es)\b[^.]{{0,120}}{c})|(?:{c}[^.]{{0,60}}\bdeleted\b)",
+                re.I,
+            ),
+            "",
+        ),
+        (
+            "crosswalk",
+            re.compile(
+                rf"(?:\bcrosswalk\w*\b[^.]{{0,160}}{c})|(?:{c}[^.]{{0,160}}\bcrosswalk)",
+                re.I,
+            ),
+            "",
+        ),
+        (
+            "bundled",
+            re.compile(rf"{c}[^.]{{0,120}}\b(?:bundled|bundled status)\b", re.I),
+            "",
+        ),
+        (
+            "telehealth_list",
+            re.compile(
+                rf"(?:Medicare telehealth services list[^.]{{0,200}}{c})"
+                rf"|(?:{c}[^.]{{0,200}}telehealth services list)",
+                re.I,
+            ),
+            "",
+        ),
+    ]
+
+
+_RE_RANGE = re.compile(r"\b(\d{5})-(\d{5})\b")
+_RE_LETTER_RANGE = re.compile(r"\b([A-Z])(\d{4})-\1(\d{4})\b")
+
+
+def _expand_ranges(text: str) -> str:
+    """ "99441-99443" -> "99441 99442 99443" and "G0008-G0010" -> "G0008
+    G0009 G0010" so a range mentions each code."""
+
+    def rep(m: re.Match[str]) -> str:
+        a, b = int(m.group(1)), int(m.group(2))
+        return (
+            " ".join(str(n) for n in range(a, b + 1)) if 0 < b - a <= 20 else m.group(0)
+        )
+
+    def rep_letter(m: re.Match[str]) -> str:
+        letter = m.group(1)
+        a, b = int(m.group(2)), int(m.group(3))
+        return (
+            " ".join(f"{letter}{n:04d}" for n in range(a, b + 1))
+            if 0 < b - a <= 20
+            else m.group(0)
+        )
+
+    text = _RE_RANGE.sub(rep, text)
+    text = _RE_LETTER_RANGE.sub(rep_letter, text)
+    return text
+
+
+def fr_events(store: Any, code: str) -> list[EventRow]:
+    code = code.upper()
+    con = store._con()
+    rows = con.execute(
+        "SELECT a.item_key, a.p_id, a.page, a.text, i.title, i.date_published "
+        "FROM fr_anchors a JOIN items i ON i.key = a.item_key WHERE a.text LIKE ? OR a.text LIKE ?",
+        (f"%{code}%", f"%{code[:2]}___-{code[:2]}___%"),
+    ).fetchall()
+    patterns = _patterns(code)
+    seen: set[tuple[int, str, str, int]] = set()
+    out: list[EventRow] = []
+    for r in rows:
+        text = _expand_ranges(r["text"])
+        # An FR page citation's number ("91 FR 99490") is not a mention of
+        # the code — strip it before both the presence check and every
+        # event pattern below (M1: a "crosswalk ... 91 FR 99490" sentence
+        # used to read as a crosswalk event for 99490).
+        text = FR_CITE_RE.sub(" ", text)
+        if code not in text.upper():
+            continue
+        year = rule_year_of(r["title"], r["date_published"])
+        others = tuple(c for c in find_codes(text) if c != code)
+        for kind, pat, direction in patterns:
+            m = pat.search(text)
+            if not m:
+                continue
+            frm = ""
+            to = ""
+            if direction == "from":
+                # "replaces": the replaced code(s) are named after the
+                # trigger phrase, up to the end of the sentence — e.g.
+                # "...99439 (with a descriptor identical to G2058)."
+                tail_text = text[m.end() :]
+                period = tail_text.find(".")
+                if period != -1:
+                    tail_text = tail_text[:period]
+                tail_codes = tuple(tc for tc in find_codes(tail_text) if tc != code)
+                frm = " ".join(tail_codes) or " ".join(others)
+            elif direction == "to":
+                tail = find_codes(text[m.start() : m.end()])
+                to = " ".join(tc for tc in tail if tc != code) or " ".join(others)
+            # A directional kind ("replaces"/"replaced_by") with no other
+            # code found nearby is not a real relationship — e.g. "...
+            # instead of the G code" names no actual code, so the sentence
+            # is prose, not a replacement event.
+            if direction == "from" and not frm:
+                continue
+            if direction == "to" and not to:
+                continue
+            key = (year, kind, r["item_key"], r["p_id"])
+            if key in seen:
+                continue
+            seen.add(key)
+            out.append(
+                EventRow(
+                    code,
+                    year,
+                    kind,
+                    frm,
+                    to,
+                    r["item_key"],
+                    r["p_id"],
+                    r["page"],
+                    "fr",
+                    True,
+                    "",
+                )
+            )
+    return sorted(out, key=lambda e: (e.year, FR_KINDS.index(e.kind), e.p_id))
+
+
+def lineage(con: Any, store: Any, code: str) -> list[EventRow]:
+    fr = fr_events(store, code)
+    fr_years = {e.year for e in fr}
+    rvu = [
+        dataclasses.replace(ev, anchored=any(abs(ev.year - y) <= 1 for y in fr_years))
+        for ev in rvu_events(con, code)
+    ]
+    return sorted([*rvu, *fr], key=lambda e: (e.year, e.source != "fr", e.kind))

tests/cli/test_pfs_cli.py

@@ -0,0 +1,343 @@
+"""stack pfs — elements / lineage / families / review."""
+
+from __future__ import annotations
+
+import duckdb
+import pytest
+from typer.testing import CliRunner
+
+import cli.pfs as pfs_cli
+from cli import app
+from pfs.codetables import (
+    ElementRow,
+    EventRow,
+    ensure_tables,
+    read_elements,
+    read_events,
+    read_families,
+    write_elements,
+    write_events,
+)
+from pfs.extract import Extraction
+
+runner = CliRunner()
+
+
+@pytest.fixture
+def restore_families():
+    """Snapshot ``pfs.families.FAMILIES`` and restore it in teardown, even
+    if the test body raises — a test that reaches ``refresh_from`` must
+    not leave the live registry clobbered for the rest of the session."""
+    from pfs import families as mod
+
+    before = dict(mod.FAMILIES)
+    try:
+        yield mod
+    finally:
+        mod.FAMILIES.clear()
+        mod.FAMILIES.update(before)
+
+
+@pytest.fixture
+def con(monkeypatch, restore_families):
+    c = duckdb.connect(":memory:")
+    ensure_tables(c)
+    c.execute(
+        "CREATE TABLE pfs.rvu (hcpcs VARCHAR, mod VARCHAR, description VARCHAR, status_code VARCHAR, non_fac_total DOUBLE, year INTEGER)"
+    )
+    c.executemany(
+        "INSERT INTO pfs.rvu VALUES (?,?,?,?,?,?)",
+        [
+            ("99490", None, "Chrnc care mgmt staff 1st 20", "A", 1.0, 2026),
+            ("G2058", None, "Ccm add 20min", "A", 1.0, 2020),
+            ("99999", None, "bundled thing", "B", 0.0, 2026),
+        ],
+    )
+
+    class _Batch:
+        def __enter__(self):
+            return c
+
+        def __exit__(self, *a):
+            return False
+
+    monkeypatch.setattr(pfs_cli, "_batch", lambda: _Batch())
+    monkeypatch.setattr(pfs_cli, "_read", lambda: c)
+    monkeypatch.setattr(pfs_cli, "_store", lambda: object())
+    published = []
+    monkeypatch.setattr(pfs_cli, "_publish", lambda: published.append(True))
+
+    class _ConProxy:
+        """``DuckDBPyConnection`` has no ``__dict__`` (C extension type), so
+        a plain ``con.published = []`` fails; wrap it to attach the
+        publish-tracking list the tests assert on while forwarding
+        everything else (execute, executemany, ...) to the real connection."""
+
+        def __init__(self, real, published):
+            object.__setattr__(self, "_real", real)
+            object.__setattr__(self, "published", published)
+
+        def __getattr__(self, name):
+            return getattr(object.__getattribute__(self, "_real"), name)
+
+    try:
+        yield _ConProxy(c, published)
+    finally:
+        c.close()
+
+
+class TestElements:
+    def test_writes_rows_and_publishes(self, con, monkeypatch):
+        def fake_extract(store, con_, code, *, classify=None):
+            return Extraction(
+                code,
+                (
+                    ElementRow(
+                        code,
+                        2026,
+                        "activity",
+                        "consent",
+                        "",
+                        "Consent;",
+                        "K",
+                        1,
+                        2,
+                        "fr",
+                    ),
+                ),
+                (),
+            )
+
+        monkeypatch.setattr(pfs_cli, "extract_code", fake_extract)
+        res = runner.invoke(app, ["pfs", "elements", "--code", "g0556", "--no-llm"])
+        assert res.exit_code == 0, res.output
+        assert "G0556: 1 elements, 0 for review" in res.output
+        assert [r.value for r in read_elements(con, "G0556")] == ["consent"]
+        assert con.published == [True]
+
+    def test_dry_run_does_not_write(self, con, monkeypatch):
+        monkeypatch.setattr(
+            pfs_cli,
+            "extract_code",
+            lambda s, c, code, *, classify=None: Extraction(code, (), ()),
+        )
+
+        def fail_batch():
+            raise AssertionError("--dry-run must not open a RW duckdb_batch connection")
+
+        monkeypatch.setattr(pfs_cli, "_batch", fail_batch)
+
+        read_calls = []
+        orig_read = pfs_cli._read
+
+        def spy_read():
+            read_calls.append(True)
+            return orig_read()
+
+        monkeypatch.setattr(pfs_cli, "_read", spy_read)
+
+        res = runner.invoke(
+            app, ["pfs", "elements", "--code", "G0556", "--no-llm", "--dry-run"]
+        )
+        assert res.exit_code == 0, res.output
+        assert con.published == []
+        assert read_calls == [True]
+
+    def test_all_payable_selects_art_status_in_newest_year(self, con, monkeypatch):
+        seen = []
+        monkeypatch.setattr(
+            pfs_cli,
+            "extract_code",
+            lambda s, c, code, *, classify=None: (
+                seen.append(code) or Extraction(code, (), ())
+            ),
+        )
+        res = runner.invoke(
+            app, ["pfs", "elements", "--all-payable", "--no-llm", "--dry-run"]
+        )
+        assert res.exit_code == 0 and seen == ["99490"]
+
+    def test_repeated_family_option_expands_every_family(self, con, monkeypatch):
+        # #684 regression: --family used to be a single str, so a second
+        # --family silently discarded the first. It must now be repeatable
+        # and expand every family named.
+        seen = []
+        monkeypatch.setattr(
+            pfs_cli,
+            "extract_code",
+            lambda s, c, code, *, classify=None: (
+                seen.append(code) or Extraction(code, (), ())
+            ),
+        )
+        res = runner.invoke(
+            app,
+            [
+                "pfs",
+                "elements",
+                "--family",
+                "CCM",
+                "--family",
+                "APCM",
+                "--no-llm",
+                "--dry-run",
+            ],
+        )
+        assert res.exit_code == 0, res.output
+        from pfs.families import HAND_FAMILIES
+
+        expected = set(HAND_FAMILIES["CCM"].codes) | set(HAND_FAMILIES["APCM"].codes)
+        assert set(seen) == expected
+
+
+class TestLineage:
+    def test_prints_and_writes(self, con, monkeypatch):
+        ev = EventRow(
+            "G2058",
+            2021,
+            "replaced_by",
+            "",
+            "99439",
+            "YBM4IZUS",
+            1578,
+            84639,
+            "fr",
+            True,
+            "",
+        )
+        monkeypatch.setattr(pfs_cli, "lineage", lambda c, s, code: [ev])
+        res = runner.invoke(app, ["pfs", "lineage", "--code", "G2058", "--write"])
+        assert res.exit_code == 0, res.output
+        assert "2021  replaced_by" in res.output and "YBM4IZUS ¶1578" in res.output
+        assert read_events(con, "G2058")[0].to_codes == "99439"
+        assert con.published == [True]
+
+    def test_without_write_reads_and_never_takes_the_write_lock(self, con, monkeypatch):
+        # I4: `stack pfs lineage` without --write must read the replica,
+        # not open a RW duckdb_batch connection (the DuckDB single-writer
+        # lock a notebook may be holding, #508-#514).
+        def fail_batch():
+            raise AssertionError(
+                "lineage without --write must not open a RW duckdb_batch connection"
+            )
+
+        monkeypatch.setattr(pfs_cli, "_batch", fail_batch)
+        monkeypatch.setattr(pfs_cli, "lineage", lambda c, s, code: [])
+
+        read_calls = []
+        orig_read = pfs_cli._read
+
+        def spy_read():
+            read_calls.append(True)
+            return orig_read()
+
+        monkeypatch.setattr(pfs_cli, "_read", spy_read)
+
+        res = runner.invoke(app, ["pfs", "lineage", "--code", "G2058"])
+        assert res.exit_code == 0, res.output
+        assert read_calls == [True]
+        assert con.published == []
+
+
+class TestFamilies:
+    def test_derives_from_tables_and_writes(self, con):
+        write_elements(
+            con,
+            "99439",
+            [
+                ElementRow(
+                    "99439", 2021, "relation", "addon-of", "99490", "", "K", 1, 1, "fr"
+                )
+            ],
+            [],
+        )
+        write_events(
+            con,
+            "G2058",
+            [
+                EventRow(
+                    "G2058", 2021, "replaced_by", "", "99439", "K", 2, 1, "fr", True, ""
+                )
+            ],
+        )
+        res = runner.invoke(app, ["pfs", "families", "--write"])
+        assert res.exit_code == 0, res.output
+        fams = read_families(con)
+        assert {r.code for r in fams if r.key == "CCM"} >= {"99490", "99439", "G2058"}
+        assert "CCM" in res.output
+
+    def test_write_tolerates_null_rvu_description(self, con):
+        # #687 regression: a real pfs.rvu row can carry a NULL description
+        # (a corpus parsing artifact) at the current max year with a
+        # payable status — that row lands in the derivation scope and used
+        # to crash `families --write` with an AttributeError.
+        con.execute(
+            "INSERT INTO pfs.rvu VALUES (?,?,?,?,?,?)",
+            ("Z9999", None, None, "A", 1.0, 2026),
+        )
+        res = runner.invoke(app, ["pfs", "families", "--write"])
+        assert res.exit_code == 0, res.output
+
+    def test_without_write_reads_and_never_takes_the_write_lock(self, con, monkeypatch):
+        # I4: `stack pfs families` without --write must read the replica,
+        # not open a RW duckdb_batch connection.
+        def fail_batch():
+            raise AssertionError(
+                "families without --write must not open a RW duckdb_batch connection"
+            )
+
+        monkeypatch.setattr(pfs_cli, "_batch", fail_batch)
+
+        read_calls = []
+        orig_read = pfs_cli._read
+
+        def spy_read():
+            read_calls.append(True)
+            return orig_read()
+
+        monkeypatch.setattr(pfs_cli, "_read", spy_read)
+
+        res = runner.invoke(app, ["pfs", "families"])
+        assert res.exit_code == 0, res.output
+        assert read_calls == [True]
+        assert con.published == []
+
+    def test_no_derived_tables_yet_is_a_friendly_no_op(self, monkeypatch):
+        # I4: a fresh replica with no pfs.code_element/code_event/code_family
+        # (nothing has run --write yet) is not a bug — print a friendly
+        # message 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", "families"])
+            assert res.exit_code == 0, res.output
+            assert "no derived tables yet" in res.output
+        finally:
+            bare.close()
+
+    def test_unknown_family_error_lists_hand_families_only(self, con, monkeypatch):
+        # M3: FAMILIES can hold ~15k derived keys once `refresh_from` has
+        # run — the error text must stay short and human-scannable.
+        pfs_cli.FAMILIES["ZZZZ-99999"] = pfs_cli.FAMILIES["CCM"]
+        res = runner.invoke(
+            app, ["pfs", "elements", "--family", "NOPE", "--no-llm", "--dry-run"]
+        )
+        assert res.exit_code != 0
+        assert "ZZZZ-99999" not in res.output
+        for key in pfs_cli.HAND_FAMILIES:
+            assert key in res.output
+
+
+class TestReview:
+    def test_lists_queue(self, con):
+        from pfs.codetables import ReviewRow
+
+        write_elements(
+            con, "G0556", [], [ReviewRow("G0556", "Odd line;", "", "", "K", 9)]
+        )
+        res = runner.invoke(app, ["pfs", "review"])
+        assert res.exit_code == 0 and "G0556" in res.output and "Odd line" in res.output

tests/llm/test_classify.py

@@ -0,0 +1,106 @@
+"""llm.classify — one-of-N choice from the local model, never free text."""
+
+from __future__ import annotations
+
+from llm.classify import build_prompt, closed_vocab_classifier, parse_choice
+
+
+class TestParseChoice:
+    def test_exact(self):
+        assert parse_choice("consent", ["consent", "24-7-access"]) == "consent"
+
+    def test_quoted_and_cased(self):
+        assert (
+            parse_choice('  "24-7-Access".\n', ["consent", "24-7-access"])
+            == "24-7-access"
+        )
+
+    def test_none_and_garbage(self):
+        assert parse_choice("none", ["consent"]) is None
+        assert (
+            parse_choice(
+                "I think it is about consent and access", ["consent", "24-7-access"]
+            )
+            is None
+        )
+
+
+class TestPrompt:
+    def test_prompt_lists_choices_and_none(self):
+        msgs = build_prompt(
+            "Provide 24/7 access for urgent needs", ["consent", "24-7-access"]
+        )
+        assert msgs[0]["role"] == "system"
+        user = msgs[1]["content"]
+        assert "consent" in user and "24-7-access" in user and "none" in user
+        assert "Provide 24/7 access" in user
+
+
+class _Pool:
+    def __init__(self):
+        self.checked = []
+
+    def check(self, model):
+        self.checked.append(model)
+        return ["http://h"]
+
+    def vram(self, host):
+        return 0.0
+
+    def serves(self, host, model):
+        return True
+
+    class _Ctx:
+        def __enter__(self):
+            return "http://h"
+
+        def __exit__(self, *a):
+            return False
+
+    def acquire_generation(self):
+        return self._Ctx()
+
+
+class _Cfg:
+    instruct_model = "qwen2.5:14b"
+    instruct_model_large = ""
+    large_min_vram_gb = 20.0
+    chat_num_ctx = 8192
+    host_vram = {}
+
+
+class TestClassifier:
+    def test_returns_choice_from_model_reply(self):
+        calls = []
+
+        def post(url, json):
+            calls.append((url, json))
+            return {"message": {"content": "24-7-access"}}
+
+        classify = closed_vocab_classifier(_Cfg(), _Pool(), post=post)
+        assert (
+            classify("Provide 24/7 access for urgent needs", ["consent", "24-7-access"])
+            == "24-7-access"
+        )
+        url, body = calls[0]
+        assert url == "http://h/api/chat" and body["stream"] is False
+        assert body["options"]["temperature"] == 0
+
+    def test_unparseable_reply_is_none(self):
+        classify = closed_vocab_classifier(
+            _Cfg(),
+            _Pool(),
+            post=lambda url, json: {"message": {"content": "maybe consent?"}},
+        )
+        assert classify("x", ["consent"]) is None
+
+    def test_pool_check_runs_once_not_per_call(self):
+        # I5: pool.check is a host-liveness probe, done once when the
+        # classifier closure is built — not per classify() call.
+        pool = _Pool()
+        classify = closed_vocab_classifier(
+            _Cfg(), pool, post=lambda url, json: {"message": {"content": "consent"}}
+        )
+        classify("a", ["consent"])
+        classify("b", ["consent"])
+        assert pool.checked == ["qwen2.5:14b"]

tests/pfs/test_codetables.py

@@ -0,0 +1,181 @@
+"""pfs.codetables — DDL + writers for the code element/event/family tables."""
+
+from __future__ import annotations
+
+import duckdb
+import pytest
+
+from pfs.codetables import (
+    ElementRow,
+    EventRow,
+    FamilyRow,
+    ReviewRow,
+    ensure_tables,
+    read_all_elements,
+    read_all_events,
+    read_elements,
+    read_events,
+    read_families,
+    write_elements,
+    write_events,
+    write_families,
+)
+
+
+@pytest.fixture
+def con():
+    c = duckdb.connect(":memory:")
+    ensure_tables(c)
+    yield c
+    c.close()
+
+
+def _el(code="99490", value="consent", year=2015):
+    return ElementRow(
+        code, year, "activity", value, "", "Consent;", "DE2VH9PD", 1245, 67716, "fr"
+    )
+
+
+class TestDDL:
+    def test_idempotent(self, con):
+        ensure_tables(con)
+        names = {
+            r[0]
+            for r in con.execute(
+                "SELECT table_name FROM information_schema.tables WHERE table_schema='pfs'"
+            ).fetchall()
+        }
+        assert {
+            "code_element",
+            "code_element_review",
+            "code_event",
+            "code_family",
+        } <= names
+
+
+class TestElements:
+    def test_write_replaces_per_code(self, con):
+        assert (
+            write_elements(
+                con,
+                "99490",
+                [_el(), _el(value="24-7-access")],
+                [ReviewRow("99490", "odd line", "activity", "", "DE2VH9PD", 1246)],
+            )
+            == 2
+        )
+        assert write_elements(con, "99490", [_el()], []) == 1
+        rows = read_elements(con, "99490")
+        assert [r.value for r in rows] == ["consent"]
+        assert (
+            con.execute(
+                "SELECT count(*) FROM pfs.code_element_review WHERE code='99490'"
+            ).fetchone()[0]
+            == 0
+        )
+
+    def test_other_codes_untouched(self, con):
+        write_elements(con, "99490", [_el()], [])
+        write_elements(con, "G0556", [_el(code="G0556")], [])
+        write_elements(con, "99490", [], [])
+        assert read_elements(con, "99490") == []
+        assert len(read_elements(con, "G0556")) == 1
+
+
+class TestReadAll:
+    def test_read_all_elements_matches_per_code_reader(self, con):
+        write_elements(
+            con, "99490", [_el(value="24-7-access"), _el(value="consent")], []
+        )
+        write_elements(con, "G0556", [_el(code="G0556")], [])
+        out = read_all_elements(con)
+        assert set(out) == {"99490", "G0556"}
+        assert out["99490"] == read_elements(con, "99490")
+        assert out["G0556"] == read_elements(con, "G0556")
+
+    def test_read_all_elements_empty(self, con):
+        assert read_all_elements(con) == {}
+
+    def test_read_all_events_matches_per_code_reader(self, con):
+        rows = [
+            EventRow(
+                "99439",
+                2021,
+                "replaces",
+                "G2058",
+                "99439",
+                "YBM4IZUS",
+                1578,
+                84639,
+                "fr",
+                True,
+                "",
+            ),
+            EventRow("99439", 2021, "appeared", "", "", "", 0, 0, "rvu", True, ""),
+        ]
+        write_events(con, "99439", rows)
+        write_events(
+            con,
+            "G2058",
+            [EventRow("G2058", 2020, "appeared", "", "", "", 0, 0, "rvu", True, "")],
+        )
+        out = read_all_events(con)
+        assert set(out) == {"99439", "G2058"}
+        assert out["99439"] == read_events(con, "99439")
+        assert out["G2058"] == read_events(con, "G2058")
+
+    def test_read_all_events_empty(self, con):
+        assert read_all_events(con) == {}
+
+
+class TestEvents:
+    def test_roundtrip_sorted_by_year(self, con):
+        rows = [
+            EventRow(
+                "99439",
+                2021,
+                "replaces",
+                "G2058",
+                "99439",
+                "YBM4IZUS",
+                1578,
+                84639,
+                "fr",
+                True,
+                "",
+            ),
+            EventRow("99439", 2021, "appeared", "", "", "", 0, 0, "rvu", True, ""),
+        ]
+        assert write_events(con, "99439", rows) == 2
+        got = read_events(con, "99439")
+        assert [e.kind for e in got] == ["appeared", "replaces"]
+        assert got[1].anchored is True
+
+
+class TestFamilies:
+    def test_full_replace(self, con):
+        write_families(
+            con,
+            [
+                FamilyRow(
+                    "CCM", "Chronic Care Management", "99490", "base", 2015, None, "", 0
+                )
+            ],
+        )
+        write_families(
+            con,
+            [
+                FamilyRow(
+                    "CCM",
+                    "Chronic Care Management",
+                    "99439",
+                    "add-on",
+                    2021,
+                    None,
+                    "YBM4IZUS",
+                    1578,
+                )
+            ],
+        )
+        fams = read_families(con)
+        assert len(fams) == 1 and fams[0].code == "99439" and fams[0].role == "add-on"

tests/pfs/test_descriptors.py

@@ -0,0 +1,229 @@
+"""pfs.descriptors — where descriptor text comes from."""
+
+from __future__ import annotations
+
+import sqlite3
+
+import duckdb
+import pytest
+
+from pfs.descriptors import (
+    DescriptorRun,
+    descriptor_runs,
+    hcpcs_long_description,
+    rule_year_of,
+    rvu_descriptions,
+)
+
+
+class _Store:
+    """Minimal stand-in for bib.Store: only ``_con()`` is used."""
+
+    def __init__(self) -> None:
+        self.con = sqlite3.connect(":memory:")
+        self.con.row_factory = sqlite3.Row
+        self.con.executescript(
+            """
+            CREATE TABLE items (key TEXT PRIMARY KEY, title TEXT, date_published TEXT);
+            CREATE TABLE fr_anchors (item_key TEXT, p_id INTEGER, page INTEGER, ordinal INTEGER, text TEXT);
+            """
+        )
+
+    def _con(self):
+        return self.con
+
+    def close(self):
+        self.con.close()
+
+
+@pytest.fixture
+def store():
+    s = _Store()
+    s.con.execute(
+        "INSERT INTO items VALUES ('DE2VH9PD', 'Medicare Program; CY 2015 PFS Final Rule', '2014-11-13')"
+    )
+    rows = [
+        (
+            "DE2VH9PD",
+            1243,
+            67716,
+            "Comment: commenters noted the CPT panel created a code.",
+        ),
+        (
+            "DE2VH9PD",
+            1244,
+            67716,
+            "These commenters suggested that we use the new CPT code 99490 (Chronic care management services, at least 20 minutes of clinical staff time directed by a physician or other qualified health care professional, per calendar month, with the following required elements:",
+        ),
+        (
+            "DE2VH9PD",
+            1245,
+            67716,
+            "Multiple (two or more) chronic conditions expected to last at least 12 months, or until the death of the patient;",
+        ),
+        (
+            "DE2VH9PD",
+            1246,
+            67716,
+            "Chronic conditions place the patient at significant risk of death, acute exacerbation/decompensation, or functional decline;",
+        ),
+        (
+            "DE2VH9PD",
+            1247,
+            67716,
+            "Comprehensive care plan established, implemented, revised, or monitored).",
+        ),
+        (
+            "DE2VH9PD",
+            1248,
+            67716,
+            "Many of these commenters expressed a preference for the per calendar month period.",
+        ),
+        ("DE2VH9PD", 1249, 67716, "Response: It is our preference to use CPT codes."),
+    ]
+    s.con.executemany(
+        "INSERT INTO fr_anchors VALUES (?,?,?,?,?)",
+        [(k, p, pg, p, t) for k, p, pg, t in rows],
+    )
+    yield s
+    s.close()
+
+
+@pytest.fixture
+def con():
+    c = duckdb.connect(":memory:")
+    c.execute("CREATE SCHEMA terminology; CREATE SCHEMA pfs")
+    c.execute(
+        "CREATE TABLE terminology.hcpcs_level_2 (hcpcs VARCHAR, seqnum VARCHAR, recid VARCHAR, long_description VARCHAR, short_description VARCHAR)"
+    )
+    c.execute(
+        "INSERT INTO terminology.hcpcs_level_2 VALUES ('G0556','10','3','Advanced primary care management services … per calendar month','Adv prim care mgmt lvl 1')"
+    )
+    c.execute(
+        "INSERT INTO terminology.hcpcs_level_2 VALUES ('G0556','5','1','Lower seqnum description for G0556','Short desc')"
+    )
+    c.execute(
+        "CREATE TABLE pfs.rvu (hcpcs VARCHAR, mod VARCHAR, description VARCHAR, status_code VARCHAR, year INTEGER)"
+    )
+    c.executemany(
+        "INSERT INTO pfs.rvu VALUES (?,?,?,?,?)",
+        [
+            ("99490", None, "Chron care mgmt srvc 20 min", "A", 2015),
+            ("99490", "26", "ignored modifier row", "A", 2015),
+            ("99490", "", "Chrnc care mgmt staff 1st 20", "A", 2022),
+        ],
+    )
+    yield c
+    c.close()
+
+
+class TestRuleYear:
+    def test_from_title(self):
+        assert (
+            rule_year_of("Medicare Program; CY 2015 PFS Final Rule", "2014-11-13")
+            == 2015
+        )
+        assert (
+            rule_year_of(
+                "Medicare and Medicaid Programs; CY 2027 Payment Policies", "2026-07-16"
+            )
+            == 2027
+        )
+
+    def test_fallback_to_date_plus_one(self):
+        assert rule_year_of("Revisions to Payment Policies", "2017-07-21") == 2018
+
+
+class TestDescriptorRuns:
+    def test_stem_and_following_elements(self, store):
+        runs = descriptor_runs(store, "99490")
+        assert len(runs) == 1
+        run = runs[0]
+        assert isinstance(run, DescriptorRun)
+        assert run.item_key == "DE2VH9PD" and run.rule_year == 2015
+        assert run.stem.p_id == 1244
+        assert [p.p_id for p in run.elements] == [1245, 1246, 1247]
+        assert (
+            "per calendar month" in run.text and "Comprehensive care plan" in run.text
+        )
+
+    def test_no_stem_no_run(self, store):
+        assert descriptor_runs(store, "99491") == []
+
+    def test_max_elements_cap(self, store):
+        run = descriptor_runs(store, "99490", max_elements=2)[0]
+        assert [p.p_id for p in run.elements] == [1245, 1246]
+
+    def test_enumeration_list_item_is_not_a_descriptor_stem(self, store):
+        # C1 live-corpus incident: "( 9) 99439 (code for non-complex
+        # chronic care management)." reads like a stem ("99439 (") but is
+        # an enumeration/cross-reference line, not the opening of 99439's
+        # own descriptor.
+        store.con.executemany(
+            "INSERT INTO fr_anchors VALUES (?,?,?,?,?)",
+            [
+                (
+                    "DE2VH9PD",
+                    1300,
+                    67800,
+                    1300,
+                    "( 9) 99439 (code for non-complex chronic care management).",
+                ),
+                (
+                    "DE2VH9PD",
+                    1301,
+                    67800,
+                    1301,
+                    "( 10) 99457 and 99458 (codes for remote physiologic monitoring).",
+                ),
+            ],
+        )
+        assert descriptor_runs(store, "99439") == []
+
+    def test_lowercase_stem_still_matches(self):
+        # `_stem_pattern` stays case-insensitive: some rules print the
+        # descriptor lowercase.
+        s = _Store()
+        s.con.execute(
+            "INSERT INTO items VALUES ('AAAAAAAA', 'Medicare Program; CY 2021 PFS Final Rule', '2020-11-13')"
+        )
+        s.con.executemany(
+            "INSERT INTO fr_anchors VALUES (?,?,?,?,?)",
+            [
+                (
+                    "AAAAAAAA",
+                    1,
+                    100,
+                    1,
+                    "CPT code 99490 (chronic care management services, at least 20 "
+                    "minutes of clinical staff time directed by a physician or other "
+                    "qualified health care professional, per calendar month, with the "
+                    "following required elements:",
+                ),
+                ("AAAAAAAA", 2, 100, 2, "Consent;"),
+            ],
+        )
+        try:
+            runs = descriptor_runs(s, "99490")
+            assert len(runs) == 1
+        finally:
+            s.close()
+
+
+class TestDuckDBSources:
+    def test_long_description(self, con):
+        assert hcpcs_long_description(con, "G0556").startswith("Advanced primary care")
+        assert hcpcs_long_description(con, "99490") == ""
+
+    def test_long_description_deterministic_seqnum(self, con):
+        # With multiple rows for same hcpcs, ORDER BY seqnum DESC picks the highest
+        result = hcpcs_long_description(con, "G0556")
+        assert (
+            result == "Advanced primary care management services … per calendar month"
+        )
+
+    def test_rvu_descriptions_base_rows_only(self, con):
+        assert rvu_descriptions(con, "99490") == [
+            (2015, "A", "Chron care mgmt srvc 20 min"),
+            (2022, "A", "Chrnc care mgmt staff 1st 20"),
+        ]

tests/pfs/test_elements.py

@@ -0,0 +1,173 @@
+"""pfs.elements — closed element vocabulary + deterministic descriptor parse."""
+
+from __future__ import annotations
+
+from pfs.elements import (
+    VOCAB,
+    Element,
+    ElementType,
+    is_element_paragraph,
+    parse_descriptor,
+    slug,
+)
+
+CCM_99490 = (
+    "Chronic care management services, at least 20 minutes of clinical staff time "
+    "directed by a physician or other qualified health care professional, per calendar "
+    "month, with the following required elements: multiple (two or more) chronic "
+    "conditions expected to last at least 12 months, or until the death of the patient; "
+    "chronic conditions place the patient at significant risk of death, acute "
+    "exacerbation/decompensation, or functional decline; comprehensive care plan "
+    "established, implemented, revised, or monitored."
+)
+G2058 = (
+    "Chronic care management services, each additional 20 minutes of clinical staff time "
+    "directed by a physician or other qualified health care professional, per calendar month "
+    "(List separately in addition to code for primary procedure). (Use G2058 in conjunction "
+    "with 99490). (Do not report 99490, G2058 in the same calendar month as 99487, 99489, 99491)"
+)
+APCM_G0556_STEM = (
+    "Advanced primary care management services provided by clinical staff and directed by a "
+    "physician or other qualified health care professional who is responsible for all primary "
+    "care and serves as the continuing focal point for all needed health care services, per "
+    "calendar month, with the following elements, as appropriate:"
+)
+
+
+def _has(elements, type_, value, detail=None):
+    return any(
+        e.type == type_ and e.value == value and (detail is None or e.detail == detail)
+        for e in elements
+    )
+
+
+class TestVocab:
+    def test_every_type_has_values(self):
+        for t in ElementType:
+            assert VOCAB[t], t
+
+    def test_values_are_slugs(self):
+        for values in VOCAB.values():
+            for v in values:
+                assert v == v.lower() and " " not in v, v
+
+
+class TestTimeAndPeriod:
+    def test_first_20_minutes_per_calendar_month(self):
+        els = parse_descriptor(CCM_99490)
+        assert _has(els, ElementType.TIME, "first", "20")
+        assert _has(els, ElementType.PERIOD, "calendar-month")
+
+    def test_each_additional(self):
+        els = parse_descriptor(G2058)
+        assert _has(els, ElementType.TIME, "each-additional", "20")
+
+    def test_per_30_days(self):
+        els = parse_descriptor(
+            "Chronic care management, at least 20 minutes, per 30 days"
+        )
+        assert _has(els, ElementType.PERIOD, "30-days")
+
+
+class TestActorAndPopulation:
+    def test_clinical_staff_directed(self):
+        els = parse_descriptor(CCM_99490)
+        assert _has(els, ElementType.ACTOR, "clinical-staff-directed")
+
+    def test_personally_by_physician(self):
+        els = parse_descriptor(
+            "Chronic care management services, provided personally by a physician or other "
+            "qualified health care professional, at least 30 minutes, per calendar month"
+        )
+        assert _has(els, ElementType.ACTOR, "physician-or-qhp-personally")
+
+    def test_two_or_more_chronic_conditions(self):
+        els = parse_descriptor(CCM_99490)
+        assert _has(
+            els, ElementType.POPULATION, "multiple-chronic-conditions-12-months"
+        )
+        assert _has(els, ElementType.POPULATION, "significant-risk")
+
+
+class TestActivityAndRelation:
+    def test_care_plan_activity(self):
+        els = parse_descriptor(CCM_99490)
+        assert _has(els, ElementType.ACTIVITY, "comprehensive-care-plan")
+
+    def test_addon_and_exclusions(self):
+        els = parse_descriptor(G2058)
+        assert _has(els, ElementType.RELATION, "addon-of", "99490")
+        for code in ("99487", "99489", "99491"):
+            assert _has(els, ElementType.RELATION, "not-with", code)
+        assert _has(els, ElementType.BILLING, "list-separately")
+
+    def test_apcm_stem_billing_and_actor(self):
+        els = parse_descriptor(APCM_G0556_STEM)
+        assert _has(els, ElementType.ACTOR, "clinical-staff-directed")
+        assert _has(els, ElementType.PERIOD, "calendar-month")
+        assert _has(els, ElementType.BILLING, "focal-point-all-primary-care")
+
+
+class TestModality:
+    def test_telecommunications(self):
+        els = parse_descriptor(
+            "Synchronous audio-video visit for the evaluation and management of a new patient, "
+            "furnished using an interactive telecommunications system"
+        )
+        assert _has(els, ElementType.MODALITY, "interactive-telecommunications")
+        assert _has(els, ElementType.MODALITY, "audio-video")
+
+    def test_audio_only(self):
+        els = parse_descriptor(
+            "Telephone evaluation and management service, audio-only"
+        )
+        assert _has(els, ElementType.MODALITY, "audio-only")
+
+
+class TestHelpers:
+    def test_sorted_unique(self):
+        els = parse_descriptor(G2058 + " " + G2058)
+        assert list(els) == sorted(
+            set(els), key=lambda e: (e.type.value, e.value, e.detail)
+        )
+
+    def test_element_paragraph_heuristic(self):
+        assert is_element_paragraph("Consent;")
+        assert is_element_paragraph(
+            "++ Document in patient's medical record that consent was obtained."
+        )
+        assert is_element_paragraph(
+            "Comprehensive care plan established, implemented, revised, or monitored)."
+        )
+        assert not is_element_paragraph(
+            "Comment: Several commenters noted that the CPT Editorial Panel created a new code"
+        )
+        assert not is_element_paragraph(
+            "Response: It is our preference to use CPT codes unless Medicare has a programmatic need"
+        )
+        assert not is_element_paragraph(
+            "We proposed that HCPCS codes G0556 through G0558 would describe APCM services furnished per calendar month."
+        )
+
+    def test_slug(self):
+        assert slug(Element(ElementType.TIME, "first", "20")) == "time=first:20"
+        assert slug(Element(ElementType.ACTIVITY, "consent")) == "activity=consent"
+
+    def test_enumeration_list_items_are_not_element_paragraphs(self):
+        # C1: "( 9) 99439 (code for non-complex chronic care management)."
+        # is a cross-reference enumeration item, not a fragment of 99439's
+        # own descriptor, even though it ends in ")." like a real element
+        # line. The still-accepted lines below must keep passing.
+        assert not is_element_paragraph(
+            "( 9) 99439 (code for non-complex chronic care management)."
+        )
+        assert not is_element_paragraph(
+            "( 10) 99457 and 99458 (codes for remote physiologic monitoring)."
+        )
+        assert is_element_paragraph("Consent;")
+        assert is_element_paragraph(
+            "++ Document in patient's medical record that consent was obtained."
+        )
+        assert is_element_paragraph(
+            "Comprehensive care plan established, implemented, revised, or monitored)."
+        )

tests/pfs/test_extract.py

@@ -0,0 +1,165 @@
+"""pfs.extract — deterministic parse + injected classifier → rows and review queue."""
+
+from __future__ import annotations
+
+import sqlite3
+
+from pfs.descriptors import DescriptorRun, Para, descriptor_runs
+from pfs.extract import Extraction, extract_run, extract_text
+
+STEM = Para(
+    "JJ6AM5HJ",
+    1163,
+    97864,
+    "HCPCS code G0556 ( Advanced primary care management services provided by clinical staff and directed by a "
+    "physician or other qualified health care professional who is responsible for all primary care and serves as "
+    "the continuing focal point for all needed health care services, per calendar month, with the following elements, as appropriate:",
+)
+ELS = (
+    Para("JJ6AM5HJ", 1164, 97864, "Consent;"),
+    Para(
+        "JJ6AM5HJ",
+        1168,
+        97864,
+        "Provide 24/7 access for urgent needs to care team/practitioner;",
+    ),
+    Para(
+        "JJ6AM5HJ",
+        1170,
+        97864,
+        "Deliver care in alternative ways to traditional office visits to best meet the patient's needs, such as home visits and/or expanded hours;",
+    ),
+    Para(
+        "JJ6AM5HJ",
+        1185,
+        97865,
+        "Be assessed through performance measurement of primary care quality, total cost of care, and meaningful use of Certified EHR Technology).",
+    ),
+    Para(
+        "JJ6AM5HJ", 1199, 97865, "Something the vocabulary does not know about at all;"
+    ),
+)
+RUN = DescriptorRun("G0556", "JJ6AM5HJ", 2025, STEM, ELS)
+
+
+def _vals(x, type_):
+    return sorted(r.value for r in x.rows if r.type == type_)
+
+
+class TestDeterministic:
+    def test_rows_anchor_to_the_paragraph_that_contains_them(self):
+        x = extract_run(RUN)
+        assert isinstance(x, Extraction) and x.code == "G0556"
+        by_value = {r.value: r for r in x.rows}
+        assert by_value["consent"].p_id == 1164
+        assert by_value["24-7-access"].p_id == 1168
+        assert by_value["performance-measurement"].p_id == 1185
+        assert by_value["clinical-staff-directed"].p_id == 1163  # stem-level
+        assert (
+            by_value["calendar-month"].source == "fr"
+            and by_value["calendar-month"].year == 2025
+        )
+
+    def test_unknown_paragraph_goes_to_review_without_classifier(self):
+        x = extract_run(RUN)
+        assert [r.p_id for r in x.reviews] == [1199]
+        assert x.reviews[0].proposed_value == ""
+
+    def test_home_setting_from_element_line(self):
+        assert "home" in _vals(extract_run(RUN), "setting")
+
+
+class TestClassifier:
+    def test_classifier_places_unknown_line(self):
+        seen = []
+
+        def classify(text, choices):
+            seen.append((text, tuple(choices)))
+            return "community-coordination" if "vocabulary" in text else None
+
+        x = extract_run(RUN, classify=classify)
+        assert x.reviews == ()
+        row = next(r for r in x.rows if r.value == "community-coordination")
+        assert row.type == "activity" and row.p_id == 1199
+        assert seen and "consent" in seen[0][1] and "significant-risk" in seen[0][1]
+
+    def test_classifier_none_means_review_with_no_proposal(self):
+        x = extract_run(RUN, classify=lambda t, c: None)
+        assert len(x.reviews) == 1 and x.reviews[0].proposed_value == ""
+
+    def test_classifier_not_called_for_deterministic_lines(self):
+        calls = []
+        extract_run(RUN, classify=lambda t, c: calls.append(t) or None)
+        assert calls == ["Something the vocabulary does not know about at all;"]
+
+
+class TestAnchorIntegrity:
+    def test_no_row_anchored_to_a_foreign_code_paragraph(self):
+        # C1 live-corpus incident: an enumeration/cross-reference line for
+        # a *different* code ("( 9) 99457 and 99458 (codes for remote
+        # physiologic monitoring, each additional 20 minutes).") must
+        # never contribute an ElementRow to a 99439 run. The fix lives in
+        # `descriptor_runs`/`is_element_paragraph` (it stops the run
+        # before absorbing such a line) — this test drives the real
+        # `descriptor_runs` to build the run, then checks `extract_run`'s
+        # output end to end: no row may be anchored to a paragraph whose
+        # text contains a different 5-char code and none of 99439.
+        con = sqlite3.connect(":memory:")
+        con.row_factory = sqlite3.Row
+        con.executescript(
+            "CREATE TABLE items (key TEXT PRIMARY KEY, title TEXT, date_published TEXT);"
+            "CREATE TABLE fr_anchors (item_key TEXT, p_id INTEGER, page INTEGER, ordinal INTEGER, text TEXT);"
+        )
+        con.execute(
+            "INSERT INTO items VALUES ('XFGGRBDH', "
+            "'Medicare and Medicaid Programs; CY 2025 Payment Policies', '2024-07-31')"
+        )
+        rows = [
+            (
+                832,
+                61652,
+                "CPT code 99439 (Chronic care management services, each additional 20 "
+                "minutes of clinical staff time directed by a physician or other "
+                "qualified health care professional, per calendar month, with the "
+                "following required elements:",
+            ),
+            (833, 61652, "Consent;"),
+            (
+                834,
+                61652,
+                "( 9) 99457 and 99458 (codes for remote physiologic monitoring, "
+                "each additional 20 minutes).",
+            ),
+        ]
+        con.executemany(
+            "INSERT INTO fr_anchors VALUES (?,?,?,?,?)",
+            [("XFGGRBDH", p, pg, p, t) for p, pg, t in rows],
+        )
+
+        class _S:
+            def _con(self):
+                return con
+
+        try:
+            run = descriptor_runs(_S(), "99439")[0]
+            # the enumeration line for 99457/99458 never joins the run
+            assert [p.p_id for p in run.elements] == [833]
+            x = extract_run(run)
+            for r in x.rows:
+                assert "99457" not in r.text and "99458" not in r.text
+        finally:
+            con.close()
+
+
+class TestText:
+    def test_hcpcs_long_description(self):
+        x = extract_text(
+            "G0556",
+            "… per calendar month, with the following elements: consent; 24/7 access …",
+            year=2025,
+            source="hcpcs",
+        )
+        assert {r.value for r in x.rows} >= {"calendar-month", "consent", "24-7-access"}
+        assert all(
+            r.item_key == "" and r.p_id == 0 and r.source == "hcpcs" for r in x.rows
+        )

tests/pfs/test_families.py

@@ -2,7 +2,46 @@
 
 from __future__ import annotations
 
-from pfs.families import FAMILIES, Detection, detect_codes, family_of, find_codes
+import logging
+import time
+
+import duckdb
+import pytest
+
+from pfs.codetables import (
+    ElementRow,
+    EventRow,
+    FamilyRow,
+    ensure_tables,
+    write_families,
+)
+from pfs.families import (
+    FAMILIES,
+    HAND_FAMILIES,
+    Detection,
+    derive_families,
+    detect_codes,
+    family_of,
+    find_codes,
+    load_families,
+    refresh_from,
+    stem_tokens,
+)
+
+
+@pytest.fixture
+def restore_families():
+    """Snapshot ``pfs.families.FAMILIES`` and restore it in teardown, even
+    if the test body raises — a test that calls ``refresh_from`` must not
+    leave the live registry clobbered for the rest of the session."""
+    from pfs import families as mod
+
+    before = dict(mod.FAMILIES)
+    try:
+        yield mod
+    finally:
+        mod.FAMILIES.clear()
+        mod.FAMILIES.update(before)
 
 
 class TestFindCodes:
@@ -74,3 +113,383 @@ class TestDetectCodes:
         assert detect_codes("what did commenters say about telehealth?") == Detection(
             (), (), ()
         )
+
+
+def _el(code, type_, value, detail=""):
+    return ElementRow(code, 2025, type_, value, detail, "", "", 0, 0, "fr")
+
+
+def _ev(code, year, kind, frm="", to=""):
+    return EventRow(code, year, kind, frm, to, "", 0, 0, "fr", True, "")
+
+
+class TestStemTokens:
+    def test_strips_time_and_actor_noise(self):
+        assert stem_tokens("Chrnc care mgmt staff 1st 20") == frozenset(
+            {"chrnc", "care", "mgmt"}
+        )
+        assert stem_tokens("Chrnc care mgmt phys ea addl") == frozenset(
+            {"chrnc", "care", "mgmt"}
+        )
+        assert stem_tokens("Adv prim care mgmt lvl 1") == frozenset(
+            {"adv", "prim", "care", "mgmt"}
+        )
+
+    def test_none_description_yields_empty_set(self):
+        # A real pfs.rvu row can carry a NULL description (corpus parsing
+        # artifact, e.g. a garbage hcpcs value) — stem_tokens must not
+        # crash on it (#687 live-run regression: AttributeError on
+        # `None.lower()` in derive_families).
+        assert stem_tokens(None) == frozenset()
+
+
+class TestDerive:
+    def test_ccm_reproduced_with_predecessor_and_addon_roles(self):
+        # #687 ruling 16: `not-with` is an exclusion, not a membership
+        # signal, and was dropped from _LINK_RELATIONS. The `not-with`
+        # element below is now inert as a merge edge — 99487 joins CCM
+        # through the stem-Jaccard + shared "comprehensive-care-plan"
+        # activity rule instead (its stem overlaps 99490's >= 0.5).
+        elements = {
+            "99490": [
+                _el("99490", "actor", "clinical-staff-directed"),
+                _el("99490", "activity", "comprehensive-care-plan"),
+            ],
+            "99439": [
+                _el("99439", "relation", "addon-of", "99490"),
+                _el("99439", "relation", "not-with", "99487"),
+            ],
+            "99487": [_el("99487", "activity", "comprehensive-care-plan")],
+            "99489": [_el("99489", "relation", "addon-of", "99487")],
+            "99491": [
+                _el("99491", "actor", "physician-or-qhp-personally"),
+                _el("99491", "activity", "comprehensive-care-plan"),
+            ],
+            "99437": [_el("99437", "relation", "addon-of", "99491")],
+            "G2058": [_el("G2058", "relation", "addon-of", "99490")],
+            "99497": [_el("99497", "activity", "advance-directive-discussion")],
+            "99498": [_el("99498", "relation", "addon-of", "99497")],
+        }
+        events = {
+            "G2058": [
+                _ev("G2058", 2020, "appeared"),
+                _ev("G2058", 2021, "disappeared"),
+                _ev("G2058", 2021, "replaced_by", to="99439"),
+            ],
+            "99439": [
+                _ev("99439", 2021, "appeared"),
+                _ev("99439", 2021, "replaces", frm="G2058"),
+            ],
+        }
+        descriptions = {
+            "99490": "Chrnc care mgmt staff 1st 20",
+            "99439": "Chrnc care mgmt staf ea addl",
+            "99487": "Cplx chrnc care 1st 60 min",
+            "99489": "Cplx chrnc care ea addl 30",
+            "99491": "Chrnc care mgmt phys 1st 30",
+            "99437": "Chrnc care mgmt phys ea addl",
+            "G2058": "Ccm add 20min",
+            "99497": "Advncd care plan 30 min",
+            "99498": "Advncd care plan addl 30 min",
+        }
+        rows = derive_families(elements, events, descriptions)
+        ccm = {r.code: r for r in rows if r.key == "CCM"}
+        assert set(ccm) == {
+            "99490",
+            "99439",
+            "99487",
+            "99489",
+            "99491",
+            "99437",
+            "G2058",
+        }
+        assert ccm["99439"].role == "add-on" and ccm["G2058"].role == "predecessor"
+        assert ccm["G2058"].since == 2020 and ccm["G2058"].until == 2021
+        assert ccm["99490"].role == "base"
+        acp = {r.code for r in rows if r.key == "ACP"}
+        assert acp == {"99497", "99498"}
+        assert all(
+            r.name == HAND_FAMILIES[r.key].name for r in rows if r.key in HAND_FAMILIES
+        )
+
+    def test_multi_code_replaced_by_does_not_merge(self):
+        # #687 ruling 16: a `replaced_by` event naming more than one
+        # to_codes is ambiguous (a blanket "these codes are replaced by
+        # ..." sentence spanning several families in the real corpus) and
+        # must not become a merge edge.
+        elements = {"11111": [], "22222": []}
+        events = {"11111": [_ev("11111", 2021, "replaced_by", to="22222 33333")]}
+        descriptions = {"11111": "Alpha widget", "22222": "Zeta gadget"}
+        rows = derive_families(elements, events, descriptions)
+        keys = {r.code: r.key for r in rows}
+        assert keys["11111"] != keys["22222"]
+
+    def test_not_with_relation_does_not_merge(self):
+        # #687 ruling 16: `not-with` is an explicit exclusion (cannot be
+        # billed together), the opposite of family membership.
+        elements = {
+            "99490": [_el("99490", "relation", "not-with", "G0556")],
+            "G0556": [],
+        }
+        descriptions = {
+            "99490": "Chrnc care mgmt staff 1st 20",
+            "G0556": "Adv prim care mgmt lvl 1",
+        }
+        rows = derive_families(elements, {}, descriptions)
+        keys = {r.code: r.key for r in rows}
+        assert keys["99490"] != keys["G0556"]
+
+    def test_bridging_addon_chain_splits_by_hand_family_and_warns(self, caplog):
+        # #687 ruling 16 guard: a component that reaches into two or more
+        # hand families is split by a BFS seeded from each hand family's
+        # own codes, run in HAND_FAMILIES order (ACP, CCM, PCM, TCM,
+        # APCM). Here "77777" is addon-of both a CCM seed (99490) and an
+        # ACP seed (99497), joining all three into one component; ACP
+        # comes first in HAND_FAMILIES, so its BFS reaches — and claims —
+        # the bridging code before CCM's BFS runs.
+        elements = {
+            "99490": [],
+            "99497": [],
+            "77777": [
+                _el("77777", "relation", "addon-of", "99490"),
+                _el("77777", "relation", "addon-of", "99497"),
+            ],
+        }
+        with caplog.at_level(logging.WARNING, logger="pfs.families"):
+            rows = derive_families(elements, {}, {})
+        by_code = {r.code: r for r in rows}
+        assert by_code["99490"].key == "CCM"
+        assert by_code["99497"].key == "ACP"
+        assert by_code["77777"].key == "ACP"
+        assert any("spans hand families" in rec.message for rec in caplog.records), (
+            caplog.text
+        )
+
+    def test_unknown_family_gets_stem_key(self):
+        elements = {
+            "99453": [_el("99453", "activity", "device-data-review")],
+            "99454": [_el("99454", "activity", "device-data-review")],
+        }
+        rows = derive_families(
+            elements,
+            {},
+            {
+                "99453": "Rem mntr physiol param setup",
+                "99454": "Rem mntr physiol param dev supl",
+            },
+        )
+        keys = {r.key for r in rows}
+        assert len(keys) == 1 and next(iter(keys)).startswith("REM-MNTR")
+
+    def test_no_activity_elements_never_merges_on_stem_alone(self):
+        # Both codes have identical stem tokens (jaccard == 1.0, well past
+        # the 0.5 threshold) but neither has an activity element — an empty
+        # activity set must not compare equal-and-qualifying, or every
+        # activity-less code with a common description would merge with
+        # every other one. Word order differs so a stem-based merge would
+        # be visible as a shared key even though the codes never touch.
+        elements = {"11111": [], "22222": []}
+        descriptions = {"11111": "Foo bar widget", "22222": "Bar widget foo"}
+        rows = derive_families(elements, {}, descriptions)
+        keys = {r.code: r.key for r in rows}
+        assert keys["11111"] != keys["22222"]
+
+    def test_disjoint_stem_never_merges(self):
+        # 10000/10001 share enough stem tokens (and a matching activity) to
+        # merge; 20000's stem shares no token with either, so it must never
+        # even be compared, let alone merged.
+        elements = {
+            "10000": [_el("10000", "activity", "act-a")],
+            "10001": [_el("10001", "activity", "act-a")],
+            "20000": [_el("20000", "activity", "act-a")],
+        }
+        descriptions = {
+            "10000": "Alpha beta gamma",
+            "10001": "Alpha beta delta",
+            "20000": "Zeta eta theta",
+        }
+        rows = derive_families(elements, {}, descriptions)
+        keys = {r.code: r.key for r in rows}
+        assert keys["10000"] == keys["10001"]
+        assert keys["20000"] != keys["10000"]
+
+    def test_non_code_member_is_dropped(self):
+        # I3: a corpus parsing artifact like '\x1a' must never become a
+        # family of its own — reject non-code-shaped members before
+        # deriving anything.
+        elements = {
+            "99453": [_el("99453", "activity", "device-data-review")],
+            "\x1a": [_el("\x1a", "activity", "device-data-review")],
+        }
+        descriptions = {
+            "99453": "Rem mntr physiol param setup",
+            "\x1a": "corpus artifact",
+        }
+        rows = derive_families(elements, {}, descriptions)
+        assert {r.code for r in rows} == {"99453"}
+
+    def test_duplicate_derived_keys_get_representative_suffix(self):
+        # I3: 627 stem keys in the live table are shared by >= 2 unrelated
+        # components (e.g. "GENE" x13); a collision must not silently
+        # merge them under `load_families`' one-row-per-key model.
+        elements = {
+            "10001": [_el("10001", "activity", "act-a")],
+            "10002": [_el("10002", "activity", "act-a")],
+            "20001": [_el("20001", "activity", "act-b")],
+            "20002": [_el("20002", "activity", "act-b")],
+        }
+        descriptions = {
+            "10001": "Rare widget",
+            "10002": "Rare widget",
+            "20001": "Rare widget",
+            "20002": "Rare widget",
+        }
+        rows = derive_families(elements, {}, descriptions)
+        keys = {r.code: r.key for r in rows}
+        assert keys["10001"] == keys["10002"]
+        assert keys["20001"] == keys["20002"]
+        assert keys["10001"] != keys["20001"]
+        both = {keys["10001"], keys["20001"]}
+        assert "RARE-WIDGET" in both
+        other = (both - {"RARE-WIDGET"}).pop()
+        assert other.startswith("RARE-WIDGET-")
+
+    def test_none_description_does_not_raise_and_stands_alone(self):
+        # #687 live-run regression: a code whose `descriptions` mapping
+        # holds an explicit None (a real NULL pfs.rvu description, not a
+        # missing key — dict.get(c, "") only substitutes its default when
+        # c is absent) must not crash derive_families, and with nothing
+        # else linking it to another code it lands in its own family.
+        elements = {"99999": [_el("99999", "activity", "act-a")]}
+        descriptions = {"99999": None}
+        rows = derive_families(elements, {}, descriptions)
+        assert {r.code for r in rows} == {"99999"}
+        assert rows[0].key == "99999"
+        assert rows[0].name == "99999"
+
+    def test_many_distinct_stems_is_fast(self):
+        # ~2,000 codes whose descriptions share no stem token with any
+        # other code's. The token-bucketed merge must not degrade to the
+        # O(n^2) all-pairs scan this guards against; kept generous (< 2s)
+        # so CI timing noise doesn't make it flaky.
+        def _word(i: int) -> str:
+            letters = []
+            n = i + 1
+            while n:
+                n, r = divmod(n - 1, 26)
+                letters.append(chr(97 + r))
+            return "".join(reversed(letters))
+
+        n = 2000
+        elements = {}
+        descriptions = {}
+        for i in range(n):
+            code = f"{10000 + i}"
+            elements[code] = [_el(code, "activity", "act")]
+            descriptions[code] = f"stem{_word(i)} term{_word(i + n)}"
+
+        start = time.perf_counter()
+        rows = derive_families(elements, {}, descriptions)
+        elapsed = time.perf_counter() - start
+
+        assert len(rows) == n
+        assert elapsed < 2.0
+
+
+class TestLoadAndRefresh:
+    def test_load_and_refresh_in_place(self, restore_families):
+        mod = restore_families
+        con = duckdb.connect(":memory:")
+        try:
+            ensure_tables(con)
+            assert load_families(con) == {}
+            before = dict(mod.FAMILIES)
+            assert refresh_from(con) == 0 and mod.FAMILIES == before
+            write_families(
+                con,
+                [
+                    FamilyRow(
+                        "CCM",
+                        "Chronic Care Management",
+                        "99490",
+                        "base",
+                        None,
+                        None,
+                        "",
+                        0,
+                    ),
+                    FamilyRow(
+                        "CCM",
+                        "Chronic Care Management",
+                        "G2058",
+                        "predecessor",
+                        2020,
+                        2021,
+                        "",
+                        0,
+                    ),
+                ],
+            )
+            n = refresh_from(con)
+            # Ruling 17: a hand key the table reaches (CCM) is replaced by
+            # the derived (superset) codes; every other hand key survives
+            # untouched since `refresh_from` seeds from HAND_FAMILIES
+            # first instead of clearing the registry outright.
+            assert n == len(HAND_FAMILIES)
+            assert set(mod.FAMILIES) >= set(HAND_FAMILIES)
+            assert set(mod.FAMILIES["CCM"].codes) == {"99490", "G2058"}
+            assert mod.FAMILIES["CCM"].name == HAND_FAMILIES["CCM"].name
+        finally:
+            con.close()
+
+    def test_hand_key_absent_from_table_survives(self, restore_families):
+        # Ruling 17: only CCM appears in the derived table; the other four
+        # hand families (ACP, PCM, TCM, APCM) must come through unchanged.
+        mod = restore_families
+        con = duckdb.connect(":memory:")
+        try:
+            ensure_tables(con)
+            write_families(
+                con,
+                [
+                    FamilyRow(
+                        "CCM",
+                        "Chronic Care Management",
+                        "99490",
+                        "base",
+                        None,
+                        None,
+                        "",
+                        0,
+                    )
+                ],
+            )
+            refresh_from(con)
+            for key in ("ACP", "PCM", "TCM", "APCM"):
+                assert mod.FAMILIES[key] == HAND_FAMILIES[key]
+        finally:
+            con.close()
+
+    def test_schema_present_table_absent_returns_empty(self):
+        # pfs schema created (e.g. by an earlier ensure_tables call for a
+        # sibling table) but pfs.code_family itself never materialized —
+        # an old replica shape, not a real error.
+        con = duckdb.connect(":memory:")
+        try:
+            con.execute("CREATE SCHEMA IF NOT EXISTS pfs;")
+            assert load_families(con) == {}
+        finally:
+            con.close()
+
+    def test_wrong_shape_table_raises(self):
+        # A pfs.code_family table exists but not in the expected shape —
+        # this is a real bug, not an absent table, and must not be
+        # swallowed.
+        con = duckdb.connect(":memory:")
+        try:
+            con.execute("CREATE SCHEMA IF NOT EXISTS pfs;")
+            con.execute("CREATE TABLE pfs.code_family (only_col VARCHAR);")
+            with pytest.raises(Exception):
+                load_families(con)
+        finally:
+            con.close()

tests/pfs/test_lineage.py

@@ -0,0 +1,240 @@
+"""pfs.lineage — dated events from RVU diffs and FR paragraphs, cross-checked."""
+
+from __future__ import annotations
+
+import sqlite3
+
+import duckdb
+import pytest
+
+from pfs.lineage import _expand_ranges, fr_events, lineage, rvu_events, rvu_year_span
+
+RVU_COLS = "hcpcs VARCHAR, mod VARCHAR, description VARCHAR, status_code VARCHAR, non_fac_total DOUBLE, year INTEGER"
+
+
+@pytest.fixture
+def con():
+    c = duckdb.connect(":memory:")
+    c.execute("CREATE SCHEMA pfs")
+    c.execute(f"CREATE TABLE pfs.rvu ({RVU_COLS})")
+    rows = [
+        # table spans 2015..2026 (filler code keeps the span honest)
+        *[("00000", None, "filler", "A", 1.0, y) for y in range(2015, 2027)],
+        # 99487: B in 2015-2016, A from 2017; description rename 2021; +25% revalue 2022
+        ("99487", None, "Cmplx chron care w/o pt vsit", "B", 0.0, 2015),
+        ("99487", None, "Cmplx chron care w/o pt vsit", "B", 0.0, 2016),
+        ("99487", None, "Cmplx chron care w/o pt vsit", "A", 2.00, 2017),
+        ("99487", None, "Cmplx chron care w/o pt vsit", "A", 2.02, 2018),
+        ("99487", None, "Cmplx chron care w/o pt vsit", "A", 2.05, 2019),
+        ("99487", None, "Cmplx chron care w/o pt vsit", "A", 2.10, 2020),
+        ("99487", None, "Cplx chrnc care 1st 60 min", "A", 2.12, 2021),
+        ("99487", None, "Cplx chrnc care 1st 60 min", "A", 2.65, 2022),
+        *[
+            ("99487", None, "Cplx chrnc care 1st 60 min", "A", 2.65, y)
+            for y in range(2023, 2027)
+        ],
+        ("99487", "26", "modifier row ignored", "A", 99.0, 2022),
+        # G2058: 2020 only
+        ("G2058", None, "Ccm add 20min", "A", 1.0, 2020),
+        # 99439: 2021 onward
+        *[
+            ("99439", "", "Chrnc care mgmt svc ea addl", "A", 1.0, y)
+            for y in range(2021, 2027)
+        ],
+    ]
+    c.executemany("INSERT INTO pfs.rvu VALUES (?,?,?,?,?,?)", rows)
+    yield c
+    c.close()
+
+
+class TestSpan:
+    def test_span(self, con):
+        assert rvu_year_span(con) == (2015, 2026)
+
+
+class TestRvuEvents:
+    def test_status_descriptor_and_revalue(self, con):
+        ev = rvu_events(con, "99487")
+        kinds = [(e.year, e.kind, e.note) for e in ev]
+        assert (2017, "status_change", "B→A") in kinds
+        assert (
+            2021,
+            "descriptor_change",
+            "Cmplx chron care w/o pt vsit → Cplx chrnc care 1st 60 min",
+        ) in kinds
+        assert (2022, "revalued", "+25.0%") in kinds
+        assert not any(
+            k == "appeared" for _, k, _ in kinds
+        )  # present from the first table year
+        assert all(
+            e.source == "rvu" and e.anchored is False and e.code == "99487" for e in ev
+        )
+
+    def test_appeared_and_disappeared(self, con):
+        ev = rvu_events(con, "G2058")
+        assert [(e.year, e.kind) for e in ev] == [
+            (2020, "appeared"),
+            (2021, "disappeared"),
+        ]
+
+    def test_appeared_only(self, con):
+        ev = rvu_events(con, "99439")
+        assert [(e.year, e.kind) for e in ev] == [(2021, "appeared")]
+
+    def test_unknown_code(self, con):
+        assert rvu_events(con, "99999") == []
+
+
+class _Store:
+    def __init__(self):
+        self.con = sqlite3.connect(":memory:")
+        self.con.row_factory = sqlite3.Row
+        self.con.executescript(
+            "CREATE TABLE items (key TEXT PRIMARY KEY, title TEXT, date_published TEXT);"
+            "CREATE TABLE fr_anchors (item_key TEXT, p_id INTEGER, page INTEGER, ordinal INTEGER, text TEXT);"
+        )
+
+    def _con(self):
+        return self.con
+
+    def close(self):
+        self.con.close()
+
+
+@pytest.fixture
+def store():
+    s = _Store()
+    s.con.executemany(
+        "INSERT INTO items VALUES (?,?,?)",
+        [
+            ("DE2VH9PD", "Medicare Program; CY 2015 PFS Final Rule", "2014-11-13"),
+            (
+                "YBM4IZUS",
+                "Medicare Program; CY 2021 Payment Policies Under the PFS",
+                "2020-12-28",
+            ),
+            (
+                "XFGGRBDH",
+                "Medicare and Medicaid Programs; CY 2025 Payment Policies",
+                "2024-07-31",
+            ),
+        ],
+    )
+    paras = [
+        (
+            "DE2VH9PD",
+            1251,
+            67716,
+            "Accordingly, we will adopt CPT code 99490 for Medicare CCM services, effective January 1, 2015 instead of the G code.",
+        ),
+        (
+            "YBM4IZUS",
+            686,
+            84547,
+            "We also are finalizing our proposal to allow HCPCS code G2058 (which we are finalizing in this rule as new CPT code 99439, see the codes in section II.H.) to be billed concurrently with TCM.",
+        ),
+        (
+            "YBM4IZUS",
+            1578,
+            84639,
+            "At the January 2020 RUC meeting, specialty societies requested a temporary crosswalk through CY 2021 between the value established by CMS for HCPCS code G2058 and the value of new CPT code 99439 (with a descriptor identical to G2058).",
+        ),
+        (
+            "XFGGRBDH",
+            489,
+            61652,
+            "The CPT Editorial Panel also deleted three codes (99441-99443) for reporting telephone E/M services. We note that CPT codes 99441, 99442, and 99443, each are assigned provisional status on the Medicare telehealth services list, and would return to bundled status when the telehealth flexibilities expire.",
+        ),
+        (
+            "XFGGRBDH",
+            512,
+            61700,
+            "The CPT Editorial Panel also deleted three codes (90867-90869) for reporting transcranial magnetic stimulation treatment.",
+        ),
+    ]
+    s.con.executemany(
+        "INSERT INTO fr_anchors VALUES (?,?,?,?,?)",
+        [(k, p, pg, p, t) for k, p, pg, t in paras],
+    )
+    yield s
+    s.close()
+
+
+class TestFrEvents:
+    def test_adopted_cpt(self, store):
+        ev = fr_events(store, "99490")
+        assert [(e.year, e.kind, e.item_key, e.p_id) for e in ev] == [
+            (2015, "adopted_cpt", "DE2VH9PD", 1251)
+        ]
+        assert ev[0].anchored is True and ev[0].source == "fr" and ev[0].page == 67716
+        assert not any(e.kind == "replaces" for e in ev)
+
+    def test_replaced_by_and_crosswalk(self, store):
+        ev = fr_events(store, "G2058")
+        kinds = {e.kind for e in ev}
+        assert "created" not in kinds
+        assert "replaces" not in kinds
+        assert "replaced_by" in kinds and "crosswalk" in kinds
+        to_kinds = {(e.year, e.kind, e.to_codes) for e in ev}
+        assert (2021, "replaced_by", "99439") in to_kinds
+
+    def test_replaces_from_the_successor_side(self, store):
+        ev = fr_events(store, "99439")
+        kinds = {(e.year, e.kind) for e in ev}
+        assert (2021, "created") in kinds  # ¶686 "new CPT code 99439"
+        rep = [e for e in ev if e.kind == "replaces"]
+        assert rep and rep[0].from_codes == "G2058"
+
+    def test_deleted_bundled_and_telehealth(self, store):
+        kinds = {e.kind for e in fr_events(store, "99441")}
+        assert {"deleted", "bundled", "telehealth_list"} <= kinds
+
+    def test_fr_citation_page_number_is_not_a_crosswalk_mention(self, store):
+        # M1: an FR page citation ("91 FR 99490") is not a real mention of
+        # 99490 — it must be stripped before pattern matching, or a
+        # "crosswalk ... 91 FR 99490" sentence reads as a crosswalk event.
+        store.con.execute(
+            "INSERT INTO fr_anchors VALUES (?,?,?,?,?)",
+            (
+                "DE2VH9PD",
+                1500,
+                67999,
+                1500,
+                "We note the crosswalk methodology discussed at 91 FR 99490 for related codes.",
+            ),
+        )
+        ev = fr_events(store, "99490")
+        assert not any(e.kind == "crosswalk" for e in ev)
+
+    def test_deleted_via_range_prefilter_middle_member(self, store):
+        # 90868 is a middle member of a "90867-90869" range and appears
+        # nowhere else in the paragraph literally — the SQL prefilter must
+        # still surface the row for a range-block LIKE match.
+        ev = fr_events(store, "90868")
+        assert any(e.kind == "deleted" for e in ev)
+
+
+class TestExpandRanges:
+    def test_letter_range(self):
+        assert _expand_ranges("G0008-G0010") == "G0008 G0009 G0010"
+
+
+class TestCrossCheck:
+    def test_rvu_events_anchored_by_nearby_fr_event(self, con, store):
+        ev = lineage(con, store, "G2058")
+        by_kind = {e.kind: e for e in ev}
+        assert (
+            by_kind["appeared"].year == 2020 and by_kind["appeared"].anchored is True
+        )  # 2021 is within ±1
+        assert (
+            by_kind["disappeared"].year == 2021
+            and by_kind["disappeared"].anchored is True
+        )  # 2021 replaced_by
+        assert "replaced_by" in by_kind
+        assert all(
+            not e.anchored for e in lineage(con, store, "99487") if e.source == "rvu"
+        )
+
+    def test_sorted_by_year_then_source(self, con, store):
+        ev = lineage(con, store, "99439")
+        assert [e.year for e in ev] == sorted(e.year for e in ev)