fix(cli,pfs): I4/I9/I6/M3 — families/lineage read without --write, bulk element/event reads, --all-payable marked experimental, unknown-family error trimmed

I4: `stack pfs families` and `stack pfs lineage` opened a RW
`duckdb_batch` connection even without `--write`, so a plain lookup
contended for the DuckDB single-writer lock a notebook may be holding
(#508-#514). Mirror the `elements --dry-run` shape: `_read()` when
`write` is false, `_batch()` + `ensure_tables` only when true. A
replica that hasn't had `--write` run yet (no `pfs.code_element`/
`code_event`/`code_family`) prints "no derived tables yet — run with
--write" and exits 0 instead of raising.

I9: `families --write` ran `read_elements`/`read_events` once per code
(~2xN queries) inside the write lock. Added `read_all_elements`/
`read_all_events` to `pfs.codetables` — one full-table read each,
grouped in Python, same per-code ordering as the existing readers.

I6: `--all-payable` isn't viable as written (the loop isn't inverted in
this slice per the controller's ruling) — help text now says
"experimental: ~20 min of SQL before any model call" and a
`log.warning` logs the target count when it's used.

M3: the unknown-`--family` error listed every key in `FAMILIES`
(~15k once `refresh_from` has run) — list only `HAND_FAMILIES`.

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

src/cli/pfs.py

@@ -13,25 +13,37 @@ 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_elements,
-    read_events,
+    read_all_elements,
+    read_all_events,
     write_elements,
     write_events,
     write_families,
 )
 from pfs.extract import extract_code
-from pfs.families import FAMILIES, derive_families, refresh_from
+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
@@ -85,8 +97,11 @@ def _codes_for(
         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(FAMILIES)}"
+                    f"unknown family {family!r}; known: {', '.join(HAND_FAMILIES)}"
                 )
             out.extend(fam.codes)
     if all_payable:
@@ -94,6 +109,11 @@ def _codes_for(
             "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")
@@ -107,7 +127,12 @@ def elements(
         [], "--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."
+        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,
@@ -139,6 +164,14 @@ def elements(
         _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"),
@@ -148,53 +181,86 @@ def lineage_cmd(
 ) -> None:
     """Timeline of a code: RVU-file diffs and FR paragraphs, cross-checked."""
     store = _store()
-    with _batch() as con:
-        ensure_tables(con)
-        events = lineage(con, store, code.upper())
-        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}")
-        if write:
-            write_events(con, code.upper(), events)
     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."""
-    with _batch() as con:
-        ensure_tables(con)
-        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()
-        ]
-        elements_by_code = {c: read_elements(con, c) for c in codes}
-        events = {c: read_events(con, 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()
-        }
-        rows = derive_families(elements_by_code, events, descriptions)
-        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)}")
-        if write:
+    if write:
+        with _batch() as con:
+            ensure_tables(con)
+            rows = _derive_rows(con)
+            _print_families(rows)
             write_families(con, rows)
             refresh_from(con)
-    if write:
         _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()

src/pfs/codetables.py

@@ -137,6 +137,35 @@ def read_events(con: Any, code: str) -> list[EventRow]:
     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"

tests/cli/test_pfs_cli.py

@@ -211,6 +211,32 @@ class TestLineage:
         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):
@@ -251,6 +277,60 @@ class TestFamilies:
         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):

tests/pfs/test_codetables.py

@@ -11,6 +11,8 @@ from pfs.codetables import (
     FamilyRow,
     ReviewRow,
     ensure_tables,
+    read_all_elements,
+    read_all_events,
     read_elements,
     read_events,
     read_families,
@@ -80,6 +82,52 @@ class TestElements:
         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 = [