docs(plan): P49 slice 2 — anchors, crosswalk, reaction series, explainer notebook (refs #688-#690)

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

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

@@ -0,0 +1,731 @@
+# 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|comments|rules|corpus [--batch N] [--dry-run]` |
+| `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).
+8. **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).
+
+```python
+# 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):
+
+```python
+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):
+
+```python
+@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
+```
+
+```python
+@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(...)`:
+
+```python
+@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
+```
+
+```python
+@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
+```
+
+```python
+@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:
+
+```python
+@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**
+
+```bash
+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:
+  ```python
+  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**
+
+```python
+# 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**
+
+```python
+# 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`:
+
+```python
+    "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:**
+```python
+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`.
+
+```python
+# 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**
+
+```python
+# 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:**
+```python
+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:**
+```python
+@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:**
+```python
+@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.