stack/docs/superpowers/plans/2026-09-09-code-family-anchors.md

Code-Family Anchors, Crosswalk, Reaction + Explainer Notebook (P49 slice 2: #688–#690) Implementation Plan

For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (- [ ]) syntax for tracking.

Goal: Every chunk and item in the corpus carries codes/families/elements anchors without re-embedding; each family knows its IOM and CFR sections; each family has a per-docket public-reaction series; and one marimo notebook walks a reader through every capability of P49 step by step on live data.

Architecture: Slice 1's pure modules (pfs.elements, pfs.families, pfs.lineage, llm.classify) are reused, never re-implemented. Anchors are computed once from chunk text and written two ways: a metadata-only restamp on pgvector (SQL UPDATE … cmetadata || jsonb) and code:/family: item tags in bib aggregated from that metadata. The crosswalk and reaction series are two more DuckDB tables in pfs, written through duckdb_batch and republished, exactly like slice 1. The notebook reads the replica and bib only, and degrades to a "not built yet" note per section when a table is absent.

Tech Stack: Python 3.13, DuckDB (replica reads, duckdb_batch writes), SQLite bib.Store, pgvector via SQLAlchemy text() (the llm engine), Ollama through llm.classify, marimo 0.23 + polars + Altair, typer, pytest.

Spec: docs/superpowers/specs/2026-09-09-code-family-longitudinal-design.md (§Decisions 4, 7; §Components bib/tag.py + bib/sync.py, llm/chunk.py + llm/index.py, bib/reaction.py; §Components crosswalk is issue #689). Slice 1 rulings in docs/superpowers/plans/2026-09-09-code-family-foundation.md apply (no DuckDB at import in pfs.*, hand families are a minimum, self-hosted inference only).

Global Constraints

• Self-hosted inference only: stance classification goes through llm.classify.closed_vocab_classifier; no cloud API.
• Never re-embed to add metadata: anchors are backfilled with UPDATE langchain_pg_embedding SET cmetadata = cmetadata || :patch::jsonb WHERE id = :id; --force re-index is not part of this slice.
• Chunk metadata stays the all-string dict convention: codes, families, elements are space-joined, sorted, unique strings ("" when none).
• Item tags are additive: code:<HCPCS> and family:<KEY> are added with Store.add_tag, never removed by this slice (the nightly Zotero sync is an additive union).
• DuckDB writes only inside conf.connect.duckdb_batch("aco"); conf.connect.publish_replica("aco") after every write command; read-only commands and notebooks open the replica.
• The notebook must import inside the notebooks container: src/ is on PYTHONPATH; llm (pgvector) access is optional and guarded — the notebook must render fully with only the replica and bib.sqlite.
• Never add a Co-Authored-By trailer. Commit after every task.
• Test output pristine (-W error::ResourceWarning); every DB fixture yields and closes.

---

File map

File	Responsibility
notebooks/code_families.py (new, T1, extended T8)	Explainer notebook: one section per capability, live data, graceful degradation
src/pfs/anchors.py (new, T2)	Pure: anchor_metadata(text, *, families) -> dict[str,str] computing codes/families/elements
src/llm/chunk.py (modify, T2)	stamp families/elements beside codes
src/llm/migrate.py (modify, T2)	GIN indexes on families and elements
src/llm/restamp.py (new, T3) + src/cli/llm.py (modify)	metadata-only backfill: `stack llm restamp --collection all
src/bib/codetags.py (new, T4) + src/cli/bib.py (modify)	code:/family: item tags from pgvector metadata + fr_anchors: stack bib code-tags [--collection …] [--dry-run]
src/llm/evidence.py (modify, T5)	code_cited_sources(engine, codes, *, per_code, collections=("rules","comments","corpus"))
src/pfs/guidance.py (new, T6) + src/pfs/codetables.py (modify) + src/cli/pfs.py (modify)	CFR/IOM cross-references per family → pfs.code_guidance; stack pfs guidance --family X [--write]
src/pfs/reaction.py (new, T7) + src/pfs/codetables.py + src/cli/pfs.py	per-(family, docket) counts + stance, FR Comment/Response pairs → pfs.code_reaction; stack pfs reaction --family X [--stance-sample N] [--write]
tests: tests/pfs/test_anchors.py, tests/llm/test_chunk.py (extend), tests/llm/test_restamp.py, tests/bib/test_codetags.py, tests/llm/test_evidence.py (extend), tests/pfs/test_guidance.py, tests/pfs/test_reaction.py, tests/cli/test_pfs_cli.py (extend), tests/notebooks/test_code_families_nb.py	one per module

---

Task 1: Explainer notebook — sections 0–4 on live data (elements, extraction, lineage, families)

Files:

• Create: notebooks/code_families.py
• Test: tests/notebooks/test_code_families_nb.py (new dir with __init__.py)

Interfaces:

• Consumes: pfs.elements.{VOCAB, parse_descriptor, slug}, pfs.descriptors.descriptor_runs, pfs.codetables.{read_elements, read_events, read_families}, pfs.families.{HAND_FAMILIES, load_families}, conf.connect.{duckdb, bib, theme}, bib.frlink.md_link, conf.display.plain_years.
• Produces: a marimo app whose cells each carry a # ── N. Title ── banner; every data-dependent section wrapped so a missing table renders mo.md("_Not built yet — run stack pfs …_") instead of raising.

Design (what the reader sees, in order):

0. Why a code is a bundle of elements — prose + the three telehealth steps quoted live from fr_anchors (item 2KVJ2HKX ¶394/¶396/¶398) with FR jump links via md_link.
1. Reading a descriptor — a code picker (default 99490; options = hand-family codes); descriptor_runs(store, code) → the stem paragraph and its element paragraphs as a table (p_id, text); then parse_descriptor(run.text) → typed elements table (type, value, detail) and the closed VOCAB as a reference table.
2. What the extractor wrote — pfs.code_element rows for the picked code (element, anchor item_key ¶p_id, source) and the pfs.code_element_review lines; a one-paragraph explanation of deterministic-first, model-second, review-third.
3. Lineage — pfs.code_event rows for the picked code's whole family as an Altair timeline (x = year, y = kind, colour = source, tooltip = anchor/note), the table beneath with FR links (md_link(f"p-{p_id}", store=store, item_key=item_key) inside try/except → plain item_key ¶p_id fallback), and a callout on the two dating rules (RVU-file year vs mentioning-rule year).
4. Families — hand list vs derived (read_families filtered to the picked family; roles; since/until), and how the edges work (add-on, by-reference, single-target replacement, stem + activity) in prose with the fixture example G2058 → 99439. 5–7. Placeholders rendering "Not built yet" until T8 (anchors, guidance, reaction).
5. Provenance — the replica path and mtime, pfs.code_* row counts, cms.ingest_log filtered to pfs.rvu (as the other notebooks do).

• Step 1: Write the failing test — the test imports the notebook module, checks it defines app, that every cell function is anonymous (_), and that a headless run with the replica absent does not raise (all guards trip).

# tests/notebooks/__init__.py  (empty)

# tests/notebooks/test_code_families_nb.py
"""notebooks/code_families.py — structure and headless degradation."""

from __future__ import annotations

import ast
import importlib.util
import re
from pathlib import Path

NB = Path(__file__).resolve().parents[2] / "notebooks" / "code_families.py"


def _load():
    spec = importlib.util.spec_from_file_location("code_families_nb", NB)
    mod = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(mod)
    return mod


def test_notebook_is_a_marimo_app():
    mod = _load()
    assert mod.app.__class__.__name__ == "App"


def test_cells_are_anonymous_and_banners_present():
    src = NB.read_text()
    tree = ast.parse(src)
    names = [n.name for n in tree.body if isinstance(n, ast.FunctionDef)]
    assert names and set(names) == {"_"}
    banners = re.findall(r"# ── (\d)\. ", src)
    assert [int(b) for b in banners] == list(range(0, 9))


def test_headless_run_degrades_without_data(monkeypatch, tmp_path):
    """With no replica and no bib the guards render notes instead of raising."""
    monkeypatch.setenv("STACK_DUCKDB_REPLICA", "1")
    mod = _load()
    import conf.connect as cc

    monkeypatch.setattr(cc, "duckdb", lambda *a, **k: (_ for _ in ()).throw(FileNotFoundError("no replica")))
    monkeypatch.setattr(cc, "bib", lambda *a, **k: (_ for _ in ()).throw(FileNotFoundError("no bib")))
    result = mod.app.run()  # marimo runs all cells; mo.stop cascades are fine
    assert result is not None

• Step 2: Run it to verify it fails

Run: uv run pytest tests/notebooks/test_code_families_nb.py -q Expected: FAIL — FileNotFoundError on NB.read_text() / spec_from_file_location (notebook missing).

• Step 3: Write notebooks/code_families.py

Follow notebooks/pfs_calcs.py for the header and notebooks/cy2027_pfs_proposed_rule.py for the banner/prose style. Skeleton (every cell shown; fill prose as indicated):

import marimo

__generated_with = "0.23.13"
app = marimo.App(width="medium")


@app.cell(hide_code=True)
def _():
    import marimo as mo
    return (mo,)


@app.cell(hide_code=True)
def _(mo):
    mo.md(
        """
        # Code families as first-class objects

        A physician fee schedule code is not a number — it is a **bundle of logical elements**:
        who furnishes the service, for how long, per what period, to which patients, doing which
        activities, by which modality. This notebook walks through how the stack turns that idea
        into tables you can query and cite: elements → extraction → lineage → families → anchors
        → guidance → public reaction. Every number on this page is read live from the replica and
        the bibliography; every claim links to the Federal Register paragraph it came from.
        """
    )
    return


@app.cell(hide_code=True)
def _():
    # ── 0. Setup ──
    import altair as alt
    import polars as pl

    from conf import connect
    from conf.display import plain_years

    connect.theme()
    NOTES = {}

    def _open_replica():
        try:
            return connect.duckdb("aco", read_only=True)
        except Exception as e:  # noqa: BLE001 — degrade, never crash the page
            NOTES["replica"] = f"replica unavailable: {e}"
            return None

    def _open_bib():
        try:
            return connect.bib()
        except Exception as e:  # noqa: BLE001
            NOTES["bib"] = f"bibliography unavailable: {e}"
            return None

    con = _open_replica()
    store = _open_bib()

    def q(sql, params=()):
        if con is None:
            return pl.DataFrame()
        try:
            return con.execute(sql, list(params)).pl()
        except Exception as e:  # noqa: BLE001 — a missing table is a "not built yet"
            NOTES[sql[:40]] = str(e)
            return pl.DataFrame()

    def fr_md(item_key, p_id):
        if store is None:
            return f"{item_key} ¶{p_id}"
        try:
            from bib.frlink import md_link

            return md_link(f"p-{p_id}", store=store, item_key=item_key, text=f"{item_key} ¶{p_id}")
        except Exception:  # noqa: BLE001
            return f"{item_key} ¶{p_id}"

    def not_built(cmd):
        return f"_Not built yet — run `{cmd}` and republish the replica._"

    return NOTES, alt, con, fr_md, not_built, pl, plain_years, q, store

Then, one cell per section, in this order (prose in mo.md, data in mo.ui.table / Altair):

@app.cell(hide_code=True)
def _(fr_md, mo, store):
    # ── 0. Why a code is a bundle of elements ──
    _steps = []
    if store is not None:
        try:
            _con = store._con()  # noqa: SLF001
            for _p in (394, 396, 398):
                _row = _con.execute(
                    "SELECT text FROM fr_anchors WHERE item_key = ? AND p_id = ?", ("2KVJ2HKX", _p)
                ).fetchone()
                if _row:
                    _steps.append(f"> {_row[0][:400]}… — {fr_md('2KVJ2HKX', _p)}")
        except Exception:  # noqa: BLE001
            pass
    mo.md(
        "## 0. Why a code is a bundle of elements\n\n"
        "CMS says so itself. When it decides whether a service can be furnished by telehealth it "
        "walks three steps, and the third is literally *review the elements of the service as "
        "described by the HCPCS code* (CY2026 proposed rule, 90 FR 32389):\n\n"
        + ("\n\n".join(_steps) if _steps else "_(bibliography unavailable — quotes omitted)_")
        + "\n\nEverything below is built on that one sentence."
    )
    return

@app.cell(hide_code=True)
def _(mo):
    # ── 1. Reading a descriptor ──
    from pfs.families import HAND_FAMILIES

    _codes = sorted({c for f in HAND_FAMILIES.values() for c in f.codes})
    code_picker = mo.ui.dropdown(options=_codes, value="99490", label="Code")
    mo.vstack([mo.md("## 1. Reading a descriptor"), code_picker])
    return (code_picker,)


@app.cell(hide_code=True)
def _(code_picker, fr_md, mo, pl, store):
    from pfs.descriptors import descriptor_runs
    from pfs.elements import VOCAB, parse_descriptor

    code = code_picker.value
    runs = descriptor_runs(store, code) if store is not None else []
    if not runs:
        _view = mo.md(f"_No Federal Register descriptor run found for {code} (bibliography unavailable or code never printed as a stem)._")
        elements = ()
    else:
        run = runs[-1]  # the latest rule that prints the descriptor
        _paras = pl.DataFrame(
            {
                "p_id": [run.stem.p_id, *[p.p_id for p in run.elements]],
                "role": ["stem", *["element"] * len(run.elements)],
                "text": [run.stem.text[:300], *[p.text[:300] for p in run.elements]],
            }
        )
        elements = parse_descriptor(run.text)
        _els = pl.DataFrame(
            {"type": [e.type.value for e in elements], "value": [e.value for e in elements], "detail": [e.detail for e in elements]}
        )
        _vocab = pl.DataFrame({"type": [t.value for t in VOCAB for _ in VOCAB[t]], "value": [v for t in VOCAB for v in VOCAB[t]]})
        _view = mo.vstack(
            [
                mo.md(
                    f"**{code}** as printed in {run.item_key} (CY{run.rule_year}), stem {fr_md(run.item_key, run.stem.p_id)}: "
                    "the stem paragraph opens the descriptor and each following paragraph is one element."
                ),
                mo.ui.table(_paras, label="Descriptor paragraphs"),
                mo.md("The deterministic parser reads what a regex can read — minutes, periods, code references, and the recurring phrases:"),
                mo.ui.table(_els, label="Typed elements"),
                mo.accordion({"The closed vocabulary (values grow only by review)": mo.ui.table(_vocab)}),
            ]
        )
    _view
    return code, elements

Sections 2, 3, 4 read the replica via q(...):

@app.cell(hide_code=True)
def _(code, mo, not_built, q):
    # ── 2. What the extractor wrote ──
    _els = q("SELECT type, value, detail, source, item_key, p_id, page FROM pfs.code_element WHERE code = ? ORDER BY type, value", (code,))
    _rev = q("SELECT text, proposed_value, item_key, p_id FROM pfs.code_element_review WHERE code = ? ORDER BY p_id", (code,))
    if _els.is_empty():
        _view = mo.md("## 2. What the extractor wrote\n\n" + not_built(f"stack pfs elements --code {code}"))
    else:
        _view = mo.vstack(
            [
                mo.md(
                    "## 2. What the extractor wrote\n\n"
                    "Three passes, in order: the regex parser, then the local model choosing **one slug from the closed list or `none`**, "
                    "then a review queue for what neither could place. Nothing enters `pfs.code_element` unless it is in the vocabulary."
                ),
                mo.ui.table(_els, label=f"pfs.code_element — {code} ({_els.height} rows)"),
                mo.ui.table(_rev, label=f"pfs.code_element_review — {code} ({_rev.height} lines)") if not _rev.is_empty() else mo.md("_Review queue empty for this code._"),
            ]
        )
    _view
    return

@app.cell(hide_code=True)
def _(alt, code, fr_md, mo, not_built, plain_years, q):
    # ── 3. Lineage ──
    from pfs.families import HAND_FAMILIES, family_of

    _fam = family_of(code)
    _codes = list(_fam.codes) if _fam else [code]
    _ev = q(
        "SELECT code, year, kind, from_codes, to_codes, source, anchored, item_key, p_id, note FROM pfs.code_event "
        "WHERE code IN (" + ",".join("?" * len(_codes)) + ") ORDER BY year, code, kind",
        _codes,
    )
    if _ev.is_empty():
        _view = mo.md("## 3. Lineage\n\n" + not_built(f"stack pfs lineage --code {code} --write"))
    else:
        _chart = (
            alt.Chart(_ev.to_pandas())
            .mark_circle(size=90)
            .encode(
                x=alt.X("year:O", title="Rule year"),
                y=alt.Y("kind:N", title=None),
                color=alt.Color("source:N", title="Source"),
                shape=alt.Shape("anchored:N", title="Anchored"),
                tooltip=["code", "year", "kind", "from_codes", "to_codes", "item_key", "p_id", "note"],
            )
            .properties(height=260, width=640)
        )
        _links = _ev.with_columns(
            anchor=_ev.select("item_key", "p_id").map_rows(lambda r: fr_md(r[0], r[1]) if r[0] else "rvu")["map"]
        )
        _view = mo.vstack(
            [
                mo.md(
                    "## 3. Lineage\n\n"
                    f"Every dated event for the **{_fam.name if _fam else code}** codes. RVU-file events (`source=rvu`) are dated by the "
                    "fee-schedule year; Federal Register events are dated by the **rule that mentions them** — a later rule recounting a "
                    "code's creation adds a later `created` row, which is why the earliest anchored event is the origin. An RVU event is "
                    "`anchored` when a Federal Register event for the same code lies within one rule year."
                ),
                mo.ui.altair_chart(_chart),
                mo.ui.table(plain_years(_links.drop("item_key", "p_id")), label="pfs.code_event"),
            ]
        )
    _view
    return

@app.cell(hide_code=True)
def _(code, mo, not_built, plain_years, q):
    # ── 4. Families ──
    from pfs.families import HAND_FAMILIES, family_of

    _hand = family_of(code)
    _key = _hand.key if _hand else ""
    _rows = q("SELECT key, name, code, role, since, until, item_key, p_id FROM pfs.code_family WHERE key = ? ORDER BY code", (_key,)) if _key else None
    if _rows is None or _rows.is_empty():
        _view = mo.md("## 4. Families\n\n" + not_built("stack pfs families --write"))
    else:
        _view = mo.vstack(
            [
                mo.md(
                    "## 4. Families\n\n"
                    "A family is a connected component over four kinds of edge: an **add-on** relation, a **defined-by-reference** relation, "
                    "a **single-target replacement** event, and **stem similarity with an identical activity set**. The hand-written registry "
                    f"is a floor, never a ceiling: **{_key}** lists {len(_hand.codes)} hand codes; the derived table shows {_rows.height}."
                ),
                mo.ui.table(plain_years(_rows), label=f"pfs.code_family — {_key}"),
                mo.md("Example: G2058 (2020 only) was replaced by CPT 99439 with an identical descriptor — a single-target `replaced_by` event, so G2058 joins CCM as a *predecessor*."),
            ]
        )
    _view
    return

Sections 5–7 are stubs in T1 (each a hidden cell with the banner and mo.md(not_built("…"))), and section 8:

@app.cell(hide_code=True)
def _(NOTES, con, mo, q):
    # ── 8. Provenance ──
    _counts = q(
        "SELECT 'code_element' t, count(*) n FROM pfs.code_element UNION ALL SELECT 'code_event', count(*) FROM pfs.code_event "
        "UNION ALL SELECT 'code_family', count(*) FROM pfs.code_family"
    )
    _log = q("SELECT * FROM cms.ingest_log WHERE table_name LIKE 'pfs.%' ORDER BY ingested_at DESC LIMIT 20")
    mo.vstack(
        [
            mo.md("## 8. Provenance"),
            mo.ui.table(_counts, label="pfs.code_* row counts") if not _counts.is_empty() else mo.md("_replica unavailable_"),
            mo.ui.table(_log, label="Ingest log — pfs.*") if not _log.is_empty() else mo.md("_no ingest log rows_"),
            mo.md("\n".join(f"- {k}: {v}" for k, v in NOTES.items()) if NOTES else "_All sources available._"),
        ]
    )
    return

(cms.ingest_log column names: check notebooks/cy2027_pfs_proposed_rule.py:1580-1591 and copy the exact query used there.)

• Step 4: Run the notebook headlessly and the test

Run: uv run marimo export session notebooks/code_families.py --no-sandbox --force-overwrite --continue-on-error -o /tmp/claude-1000/-home-kert-stack/fe33b7f6-fbbd-4b5e-a2ef-aac20e2431fc/scratchpad/code_families.session.json (with the data links present) and uv run python dev/scripts/nb_integration.py --set code_families.py if that entry point accepts a single notebook (read its --help); then uv run pytest tests/notebooks -q. Expected: no cell errors in the session JSON; sections 0–4 populated for 99490; 5–7 show "Not built yet"; test PASS.

• Step 5: Commit

git add notebooks/code_families.py tests/notebooks/__init__.py tests/notebooks/test_code_families_nb.py
git commit -m "feat(notebooks): code_families explainer — elements, extraction, lineage, families on live data (refs #684-#687)"

---

Task 2: Anchor metadata on chunks — families and elements beside codes

Files:

• Create: src/pfs/anchors.py
• Modify: src/llm/chunk.py:151,193 (both metadata dicts), src/llm/migrate.py (_HNSW_DDL)
• Test: tests/pfs/test_anchors.py, tests/llm/test_chunk.py (extend), tests/llm/test_migrate.py (extend)

Interfaces:

• Produces:

  def anchor_metadata(text: str, *, families: Mapping[str, Family] | None = None) -> dict[str, str]
      # {"codes": "99439 99490", "families": "CCM", "elements": "activity=consent time=first"}; families default = pfs.families.FAMILIES (live)
  

  elements = sorted unique f"{type}={value}" (no detail) from parse_descriptor(text); families = sorted unique keys of family_of(code) for each code.

• Step 1: Failing tests

# tests/pfs/test_anchors.py
from pfs.anchors import anchor_metadata
from pfs.families import Family


def test_codes_families_elements():
    md = anchor_metadata("Use 99439 in conjunction with 99490; consent; per calendar month")
    assert md["codes"] == "99439 99490"
    assert md["families"] == "CCM"
    assert "activity=consent" in md["elements"].split() and "period=calendar-month" in md["elements"].split()
    assert "relation=addon-of" in md["elements"].split()  # detail dropped


def test_empty_and_unknown():
    assert anchor_metadata("nothing here") == {"codes": "", "families": "", "elements": ""}
    fams = {"X": Family("X", "X fam", ("12345",), ("x fam",))}
    assert anchor_metadata("code 12345", families=fams)["families"] == "X"

Extend tests/llm/test_chunk.py: for a paragraph doc and a plain doc, every chunk's metadata has keys codes, families, elements and a chunk mentioning G0556 has families == "APCM". Extend tests/llm/test_migrate.py: _HNSW_DDL contains ix_embedding_families and ix_embedding_elements GIN statements shaped like ix_embedding_codes.

• Step 2: Run to fail. uv run pytest tests/pfs/test_anchors.py tests/llm/test_chunk.py tests/llm/test_migrate.py -q → ImportError / KeyError.

• Step 3: Implement

# src/pfs/anchors.py
"""Anchor metadata for a piece of text: the codes it names, the families
those codes belong to, and the element slugs its phrases match. One
function, used by the indexer at chunk time and by the restamp backfill,
so both write identical values. Pure."""

from __future__ import annotations

from typing import Mapping

from pfs.elements import parse_descriptor
from pfs.families import FAMILIES, Family, find_codes


def anchor_metadata(text: str, *, families: Mapping[str, Family] | None = None) -> dict[str, str]:
    fams = FAMILIES if families is None else families
    codes = find_codes(text)
    keys = sorted({k for k, f in fams.items() for c in codes if c in f.codes})
    elements = sorted({f"{e.type.value}={e.value}" for e in parse_descriptor(text)})
    return {"codes": " ".join(codes), "families": " ".join(keys), "elements": " ".join(elements)}

In src/llm/chunk.py replace both "codes": " ".join(find_codes(...)) entries with **anchor_metadata(f"{heading} {piece}") / **anchor_metadata(text) (import from pfs.anchors import anchor_metadata; keep find_codes import only if still used). In src/llm/migrate.py append to _HNSW_DDL:

    "CREATE INDEX IF NOT EXISTS ix_embedding_families "
    "ON langchain_pg_embedding USING gin "
    "(string_to_array(coalesce(cmetadata->>'families',''), ' '))",
    "CREATE INDEX IF NOT EXISTS ix_embedding_elements "
    "ON langchain_pg_embedding USING gin "
    "(string_to_array(coalesce(cmetadata->>'elements',''), ' '))",

• Step 4: Run to pass, plus uv run pytest tests/llm -q.
• Step 5: Commit — feat(pfs,llm): anchor metadata — families and elements stamped beside codes; GIN indexes (refs #688).

---

Task 3: Metadata-only backfill — stack llm restamp

Files:

• Create: src/llm/restamp.py
• Modify: src/cli/llm.py (new command)
• Test: tests/llm/test_restamp.py

Interfaces:

def restamp(engine, *, collection: str, batch: int = 2000, dry_run: bool = False, families=None, log_every: int = 50_000) -> dict
    # {"scanned": n, "updated": n} — iterates id, document, cmetadata ordered by id in batches (keyset pagination on id),
    # computes anchor_metadata(document), skips rows whose three values already match, else
    # UPDATE langchain_pg_embedding SET cmetadata = cmetadata || :patch WHERE id = :id (executemany per batch, one transaction per batch)

CLI: stack llm restamp --collection all|comments|rules|corpus [--batch 2000] [--dry-run] — loads .env-dependent config the way index does, calls refresh_from on the replica when cfg.duckdb_replica exists (so derived families stamp), runs migrate/ensure_hnsw first so the new GIN indexes exist, prints per-collection counts and elapsed time.

• Step 1: Failing test — uses SQLite via SQLAlchemy? No: pgvector-specific || jsonb. Test the pure parts: plan_patches(rows) -> list[tuple[id, patch_json]] (skips unchanged) and the SQL text constants; and an integration test marked @pytest.mark.skipif(not os.environ.get("LLM_DB_PASSWORD"), reason="needs pgvector") that restamps --collection rules --batch 50 --dry-run against the live DB and asserts scanned > 0.

# tests/llm/test_restamp.py
import json
from llm.restamp import plan_patches, SELECT_SQL, UPDATE_SQL


def test_plan_skips_unchanged_and_patches_changed():
    rows = [
        ("a", "Use 99439 with 99490; consent", {"codes": "99439 99490", "families": "CCM", "elements": "activity=consent relation=addon-of"}),
        ("b", "Use 99439 with 99490; consent", {"codes": "99439 99490"}),
        ("c", "nothing", {}),
    ]
    patches = plan_patches(rows)
    ids = [p[0] for p in patches]
    assert ids == ["b", "c"]
    assert json.loads(patches[0][1])["families"] == "CCM"
    assert json.loads(patches[1][1]) == {"codes": "", "families": "", "elements": ""}


def test_sql_shapes():
    assert "ORDER BY e.id" in SELECT_SQL and ":after" in SELECT_SQL and ":batch" in SELECT_SQL
    assert "cmetadata || CAST(:patch AS jsonb)" in UPDATE_SQL

• Step 2: Run to fail.
• Step 3: Implement

# src/llm/restamp.py
"""Backfill anchor metadata on already-indexed chunks without re-embedding.

Re-indexing 1.15 M chunks costs days of GPU time; the anchors are a pure
function of the chunk text, so they are recomputed in Python and merged
into ``cmetadata`` with a jsonb concatenation. Keyset pagination on the
primary key keeps memory flat; one transaction per batch keeps the lock
short. Idempotent: rows whose three values already match are skipped.
"""

from __future__ import annotations

import json
import logging
import time
from typing import Any, Iterable, Mapping

from sqlalchemy import text

from pfs.anchors import anchor_metadata

log = logging.getLogger(__name__)

KEYS = ("codes", "families", "elements")

SELECT_SQL = (
    "SELECT e.id, e.document, e.cmetadata FROM langchain_pg_embedding e "
    "WHERE e.collection_id = (SELECT uuid FROM langchain_pg_collection WHERE name = :collection) "
    "AND e.id > :after ORDER BY e.id LIMIT :batch"
)
UPDATE_SQL = "UPDATE langchain_pg_embedding SET cmetadata = cmetadata || CAST(:patch AS jsonb) WHERE id = :id"


def plan_patches(rows: Iterable[tuple[str, str, Mapping[str, Any] | None]], *, families=None) -> list[tuple[str, str]]:
    out: list[tuple[str, str]] = []
    for id_, document, md in rows:
        md = md or {}
        want = anchor_metadata(document or "", families=families)
        if all(str(md.get(k, "")) == want[k] for k in KEYS):
            continue
        out.append((id_, json.dumps(want)))
    return out


def restamp(engine: Any, *, collection: str, batch: int = 2000, dry_run: bool = False, families=None, log_every: int = 50_000) -> dict:
    scanned = updated = 0
    after = ""
    t0 = time.perf_counter()
    while True:
        with engine.begin() as conn:
            rows = conn.execute(text(SELECT_SQL), {"collection": collection, "after": after, "batch": batch}).fetchall()
            if not rows:
                break
            patches = plan_patches(rows, families=families)
            if patches and not dry_run:
                conn.execute(text(UPDATE_SQL), [{"id": i, "patch": p} for i, p in patches])
        scanned += len(rows)
        updated += len(patches)
        after = rows[-1][0]
        if scanned % log_every < batch:
            log.info("restamp %s: scanned %d updated %d (%.0fs)", collection, scanned, updated, time.perf_counter() - t0)
    return {"collection": collection, "scanned": scanned, "updated": updated, "seconds": round(time.perf_counter() - t0, 1)}

CLI in src/cli/llm.py (mirror index for config/engine setup; _COLLECTIONS fan-out for all; call migrate(engine) and ensure_hnsw(engine, cfg.embed_dim) before restamping; refresh_from guarded by os.path.exists(cfg.duckdb_replica)).

• Step 4: Run to pass; then live: bash <withenv> uv run stack llm restamp --collection rules (67k chunks, minutes), then --collection corpus, then --collection comments (918k — run in background with nohup, expect ~20–40 min; record the counts). Verify: SELECT count(*) FROM langchain_pg_embedding WHERE cmetadata ? 'families' equals the total.
• Step 5: Commit — feat(llm): stack llm restamp — metadata-only anchor backfill on pgvector (refs #688).

---

Task 4: Item tags code: / family: in bib — stack bib code-tags

Files:

• Create: src/bib/codetags.py
• Modify: src/bib/tag.py (docstring + Tag.code, Tag.family factories), src/cli/bib.py
• Test: tests/bib/test_codetags.py

Interfaces:

def item_codes_from_chunks(engine, *, collection: str) -> dict[str, tuple[set[str], set[str]]]   # item_key -> (codes, families) via GROUP BY on cmetadata->>'item_key'
def item_codes_from_anchors(store) -> dict[str, tuple[set[str], set[str]]]                      # rules: fr_anchors.text per item_key via anchor_metadata (no pgvector needed)
def apply_code_tags(store, mapping, *, dry_run=False, max_codes_per_item=200) -> dict           # {"items": n, "tags_added": n, "skipped_wide": n}

Rule: an item gets code:X for every code and family:K for every family; items with more than max_codes_per_item distinct codes (RVU files, code lists) get only family: tags plus code: for hand-family codes — a tag cloud of 8k codes helps nobody. Store.add_tag only stamps when new, so reruns are cheap.

CLI: stack bib code-tags [--collection all|comments|rules|corpus] [--anchors-only] [--dry-run].

• Step 1: Failing tests — apply_code_tags against an in-memory Store (use bib.Store(path=":memory:") if supported — check src/bib/store.py:51; else tmp_path / "bib.sqlite"): two items, one wide (300 codes) → only family tags + hand codes; item_codes_from_anchors with two fr_anchors rows; Tag.code("G0556").label == "code:G0556".
• Step 2: Run to fail.
• Step 3: Implement (SQL for chunks: SELECT cmetadata->>'item_key', string_agg(cmetadata->>'codes',' '), string_agg(cmetadata->>'families',' ') FROM langchain_pg_embedding WHERE collection_id = (…) AND coalesce(cmetadata->>'codes','') <> '' GROUP BY 1).
• Step 4: Run to pass; live: bash <withenv> uv run stack bib code-tags --collection all and uv run stack bib tag --namespace code | head (or whatever stack bib tag lists) to show counts; the nightly zotero-sync.yml carries them (no change needed — confirm by reading the workflow's --tag scopes; if it only syncs source:email/source:federal-register, add a third invocation --tag family:CCM? No — ruling: leave the nightly scopes as they are; tags ride along on items already in scope).
• Step 5: Commit — feat(bib): code:/family: item tags from chunk anchors and FR paragraphs — stack bib code-tags (refs #688).

---

Task 5: Code-cited sources across collections in the chat

Files:

• Modify: src/llm/evidence.py (code_cited_sources signature + _CITED_SQL per collection loop), src/llm/rag.py:150 call site, src/llm/config.py + stack.toml (code_cited_collections = ["rules","comments","corpus"])
• Test: tests/llm/test_evidence.py (extend), tests/llm/test_config.py (extend)

Interfaces: code_cited_sources(engine, codes, *, per_code, collections=("rules",)) -> list[dict] — runs the existing query once per collection, per-code cap applies per collection, results concatenated rules → comments → corpus; sources carry kind from metadata so llm/links.py builds the right link.

• Steps: failing test (fake engine returning rows per collection; assert order and caps), implement, run tests/llm -q, commit feat(llm): code-cited sources from comments and corpus too (refs #688).

---

Task 6: Sub-regulatory crosswalk — pfs.code_guidance + stack pfs guidance

Files:

• Create: src/pfs/guidance.py
• Modify: src/pfs/codetables.py (GuidanceRow, DDL, write_guidance, read_guidance), src/cli/pfs.py
• Test: tests/pfs/test_guidance.py, tests/cli/test_pfs_cli.py (extend)

Interfaces:

@dataclass(frozen=True) class GuidanceRow: family: str; code: str; kind: str; locator: str; item_key: str; item_key_src: str; p_id_src: int; page_src: int
    # kind ∈ {"cfr","iom","mln"}; locator = "42 CFR 410.78" | "100-04 ch.12 §30.6.x" | title; item_key = resolved bib item ("" when unresolved)
CFR_RE = re.compile(r"\b(\d{1,2})\s*CFR\s*(?:part\s*)?(\d{3,4})\.(\d+[a-z]?)", re.I)   # + "§ 410.78" with title inferred as 42
IOM_RE = re.compile(r"(?:Pub(?:lication|\.)?\s*100-0(\d)|(Claims Processing|Benefit Policy|Program Integrity) Manual)[^.]{0,80}?[Cc]hapter\s*(\d+)(?:[^.]{0,40}?[Ss]ection\s*([\d.]+))?")
def harvest(store, codes: Sequence[str], *, family: str) -> list[GuidanceRow]      # paragraphs whose text names any code → regex → resolve
def resolve_cfr(store, title: str, section: str) -> str                            # bib.cfrlink.item_for(parse_cite(f"{title} CFR {section}"), store)
def resolve_iom(store, pub: str, chapter: str) -> str                               # items tagged pub:100-0N whose title contains f"Chapter {chapter}:" (Store.list_items(tag=…))

CLI: stack pfs guidance --family CCM [--family …] [--write] (read-only without --write, same shape as families).

• Steps: failing tests (regexes on three real paragraph snippets: 2KVJ2HKX ¶398 "§ 410.78(a)(3)", a "Medicare Claims Processing Manual, Chapter 12, Section 30.6.4" sentence, a "Pub. 100-04, chapter 12" sentence; resolver with a stub store), implement, live run for the five hand families and record rows per family, commit feat(pfs): sub-regulatory crosswalk — CFR/IOM references per family with FR anchors (refs #689).

---

Task 7: Public-reaction series — pfs.code_reaction + stack pfs reaction

Files:

• Create: src/pfs/reaction.py
• Modify: src/pfs/codetables.py (ReactionRow, DDL, writers/readers), src/cli/pfs.py
• Test: tests/pfs/test_reaction.py, tests/cli/test_pfs_cli.py (extend)

Interfaces:

@dataclass(frozen=True) class ReactionRow: family: str; period: str; period_kind: str; n_items: int; n_total: int; stance_support: int; stance_oppose: int; stance_modify: int; stance_unclear: int; sample_json: str
    # period = docket id ("CMS-2025-0304") or rule item_key for FR pairs; period_kind ∈ {"docket","fr-pairs"}
STANCES = ("support", "oppose", "modify", "unclear")
def docket_counts(engine, codes) -> list[tuple[str, int, int]]                      # (docket, n_items_with_code, n_items_total) from comments chunk metadata (docket, item_key)
def stance_sample(engine, codes, docket, *, n) -> list[tuple[str, str]]             # (item_key, first chunk text mentioning a code), newest first, capped
def classify_stances(sample, classify) -> Counter                                   # closed vocab via llm.classify; prompt text = chunk text[:1500]
def fr_pairs(store, codes) -> list[tuple[str, int, int, int]]                       # (item_key, rule_year, n_comment_paras, n_response_paras) — "Comment:" paragraph followed by "Response:" where either names a code
def series(engine, store, family, codes, *, classify=None, stance_sample_n=0) -> list[ReactionRow]

CLI: stack pfs reaction --family CCM [--stance-sample 50] [--write] (stance only when --stance-sample > 0; --no-llm implied otherwise).

• Steps: failing tests (fake engine rows; fr_pairs on a sqlite _Store fixture with Comment:/Response: paragraphs; classify_stances with a fake classifier), implement, live run for CCM/APCM/G2211-family? (G2211 has no hand family — run --code G2211 too: accept --code producing a single-code "family"), commit feat(pfs): public-reaction series per family — docket counts, stance sample, FR Comment/Response pairs (refs #690).

---

Task 8: Notebook sections 5–7 on live data + docs + tracker

Files:

• Modify: notebooks/code_families.py (replace the three stubs), docs/docs/cli/*.md (regenerate)
• Test: tests/notebooks/test_code_families_nb.py (banner test already covers 0–8; add an assertion that the section-5 cell tolerates a missing LLM_DB_PASSWORD)

Section 5 Anchors in the corpus: from the replica? No — counts live in pgvector. Guarded: if LLM_DB_PASSWORD is set, query SELECT c.name, count(*) FROM langchain_pg_embedding e JOIN langchain_pg_collection c … WHERE string_to_array(cmetadata->>'families',' ') && ARRAY[:key] GROUP BY 1 and show chunks per collection for the picked family plus the item-tag counts from bib (Store.list_tags(namespace="family")); else render the bib tag counts only with a note. Section 6 Guidance: pfs.code_guidance rows for the family with CFR links (bib.cfrlink.md_link) and IOM item titles. Section 7 Reaction: pfs.code_reaction for the family as an Altair bar chart (x = period, y = n_items, tooltip = stance counts) and the FR-pairs series as a second bar chart by rule year.

• Steps: implement the three cells, run headless export (no errors), uv run pytest tests/notebooks -q, uv run python docs/scripts/extract_cli.py → commit only docs/docs/cli/{llm-restamp,bib-code-tags,pfs-guidance,pfs-reaction}.md and changed index pages, commit feat(notebooks,docs): code_families sections 5–7 — anchors, guidance, reaction; CLI docs (refs #688-#690). The controller posts the live numbers on #688–#690.

---

Self-review

Spec coverage. Decision 4 (anchors on items and chunks) → T2–T4; "code_cited_sources can pull comment and IOM chunks" (#688 done-when) → T5; #689 crosswalk with FR-paragraph provenance → T6; #690 reaction incl. pre-2017 FR pairs → T7; the notebook the user asked for → T1 + T8. stack llm index --force re-index deliberately replaced by restamp (Global Constraints). Not in this slice: IOM section-level page locating inside chapter PDFs (#689 mentions llm/pages.py; T6 resolves to the chapter item and leaves page_src for the FR anchor only — noted for the tracker), org: top-organisations column (existing org: tags can be joined in the notebook later).

Placeholders. T5–T7 give interfaces, regexes, SQL shapes and test intents but not full code bodies — the implementers are mid-tier models working from signatures, which the SDD model-selection rule allows ("prose descriptions → mid tier"). Every other step has code.

Type consistency. anchor_metadata is the single producer of codes/families/elements (T2, T3, T4); GuidanceRow/ReactionRow are defined in T6/T7 and read by T8; code_cited_sources(collections=…) in T5 matches the rag.py call site.