feat(cli): stack pfs elements|lineage|families|review (refs #684 #685 #686 #687)

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

src/cli/__init__.py

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

src/cli/pfs.py

@@ -0,0 +1,193 @@
+"""stack pfs — code elements, lineage events and derived families.
+
+    uv run stack pfs elements --code G0556 --code 99490
+    uv run stack pfs elements --family CCM
+    uv run stack pfs elements --all-payable --no-llm --dry-run
+    uv run stack pfs lineage --code G2058 [--write]
+    uv run stack pfs families [--write]
+    uv run stack pfs review [--code G0556]
+
+Writes go through ``duckdb_batch`` (single-writer rule) and republish
+the read-only replica so the chat and notebooks see them.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+import typer
+
+from pfs.codetables import (
+    ensure_tables,
+    read_elements,
+    read_events,
+    write_elements,
+    write_events,
+    write_families,
+)
+from pfs.extract import extract_code
+from pfs.families import FAMILIES, derive_families, refresh_from
+from pfs.lineage import lineage
+
+app = typer.Typer(no_args_is_help=True)
+
+
+# ── indirections the tests monkeypatch ───────────────────────────────
+def _batch() -> Any:
+    from conf.connect import duckdb_batch
+
+    return duckdb_batch("aco")
+
+
+def _read() -> Any:
+    from conf.connect import duckdb
+
+    return duckdb("aco", read_only=True)
+
+
+def _store() -> Any:
+    from conf.connect import bib
+
+    return bib()
+
+
+def _publish() -> None:
+    from conf.connect import publish_replica
+
+    typer.echo(f"replica → {publish_replica('aco')}")
+
+
+def _classifier() -> Any:
+    from llm import config as llm_config
+    from llm.classify import closed_vocab_classifier
+    from llm.pool import HostPool
+
+    cfg = llm_config.load()
+    return closed_vocab_classifier(cfg, HostPool.from_config(cfg))
+
+
+def _codes_for(con: Any, codes: list[str], family: str, all_payable: bool) -> list[str]:
+    out: list[str] = [c.upper() for c in codes]
+    if family:
+        refresh_from(con)
+        fam = FAMILIES.get(family.upper())
+        if fam is None:
+            raise typer.BadParameter(
+                f"unknown family {family!r}; known: {', '.join(FAMILIES)}"
+            )
+        out.extend(fam.codes)
+    if all_payable:
+        rows = con.execute(
+            "SELECT DISTINCT hcpcs FROM pfs.rvu WHERE status_code IN ('A','R','T') "
+            "AND year = (SELECT max(year) FROM pfs.rvu) ORDER BY hcpcs"
+        ).fetchall()
+        out.extend(r[0] for r in rows)
+    if not out:
+        raise typer.BadParameter("pass --code, --family or --all-payable")
+    return sorted(set(out))
+
+
+@app.command()
+def elements(
+    code: list[str] = typer.Option([], "--code", help="HCPCS/CPT code (repeatable)."),
+    family: str = typer.Option("", help="Expand a registered family (CCM, APCM, …)."),
+    all_payable: bool = typer.Option(
+        False, "--all-payable", help="Every A/R/T code in the newest RVU year."
+    ),
+    no_llm: bool = typer.Option(
+        False,
+        "--no-llm",
+        help="Skip the local-model classifier (unknown lines go to review).",
+    ),
+    dry_run: bool = typer.Option(
+        False, "--dry-run", help="Extract and report; write nothing."
+    ),
+) -> None:
+    """Extract typed elements for codes into pfs.code_element (+ review queue)."""
+    store = _store()
+    classify = None if no_llm else _classifier()
+    with _batch() as con:
+        ensure_tables(con)
+        targets = _codes_for(con, code, family, all_payable)
+        for c in targets:
+            x = extract_code(store, con, c, classify=classify)
+            if not dry_run:
+                write_elements(con, c, x.rows, x.reviews)
+            typer.echo(f"{c}: {len(x.rows)} elements, {len(x.reviews)} for review")
+    if not dry_run:
+        _publish()
+
+
+@app.command("lineage")
+def lineage_cmd(
+    code: str = typer.Option(..., "--code"),
+    write: bool = typer.Option(
+        False, "--write", help="Persist to pfs.code_event and republish."
+    ),
+) -> None:
+    """Timeline of a code: RVU-file diffs and FR paragraphs, cross-checked."""
+    store = _store()
+    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:
+        _publish()
+
+
+@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 = {c: read_elements(con, c) for c in codes}
+        events = {c: read_events(con, c) for c in codes}
+        descriptions = {
+            r[0]: r[1]
+            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, 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:
+            write_families(con, rows)
+            refresh_from(con)
+    if write:
+        _publish()
+
+
+@app.command()
+def review(code: str = typer.Option("", "--code")) -> None:
+    """Element lines the classifier could not place."""
+    con = _read()
+    sql = (
+        "SELECT code, text, proposed_value, item_key, p_id FROM pfs.code_element_review"
+    )
+    params: list[Any] = []
+    if code:
+        sql += " WHERE code = ?"
+        params.append(code.upper())
+    for c, text, proposed, key, p_id in con.execute(
+        sql + " ORDER BY code, p_id", params
+    ).fetchall():
+        typer.echo(f"{c}  {key} ¶{p_id}  [{proposed or '?'}]  {text[:120]}")

tests/cli/test_pfs_cli.py

@@ -0,0 +1,203 @@
+"""stack pfs — elements / lineage / families / review."""
+
+from __future__ import annotations
+
+import duckdb
+import pytest
+from typer.testing import CliRunner
+
+import cli.pfs as pfs_cli
+from cli import app
+from pfs.codetables import (
+    ElementRow,
+    EventRow,
+    ensure_tables,
+    read_elements,
+    read_events,
+    read_families,
+    write_elements,
+    write_events,
+)
+from pfs.extract import Extraction
+
+runner = CliRunner()
+
+
+@pytest.fixture
+def restore_families():
+    """Snapshot ``pfs.families.FAMILIES`` and restore it in teardown, even
+    if the test body raises — a test that reaches ``refresh_from`` must
+    not leave the live registry clobbered for the rest of the session."""
+    from pfs import families as mod
+
+    before = dict(mod.FAMILIES)
+    try:
+        yield mod
+    finally:
+        mod.FAMILIES.clear()
+        mod.FAMILIES.update(before)
+
+
+@pytest.fixture
+def con(monkeypatch, restore_families):
+    c = duckdb.connect(":memory:")
+    ensure_tables(c)
+    c.execute(
+        "CREATE TABLE pfs.rvu (hcpcs VARCHAR, mod VARCHAR, description VARCHAR, status_code VARCHAR, non_fac_total DOUBLE, year INTEGER)"
+    )
+    c.executemany(
+        "INSERT INTO pfs.rvu VALUES (?,?,?,?,?,?)",
+        [
+            ("99490", None, "Chrnc care mgmt staff 1st 20", "A", 1.0, 2026),
+            ("G2058", None, "Ccm add 20min", "A", 1.0, 2020),
+            ("99999", None, "bundled thing", "B", 0.0, 2026),
+        ],
+    )
+
+    class _Batch:
+        def __enter__(self):
+            return c
+
+        def __exit__(self, *a):
+            return False
+
+    monkeypatch.setattr(pfs_cli, "_batch", lambda: _Batch())
+    monkeypatch.setattr(pfs_cli, "_read", lambda: c)
+    monkeypatch.setattr(pfs_cli, "_store", lambda: object())
+    published = []
+    monkeypatch.setattr(pfs_cli, "_publish", lambda: published.append(True))
+
+    class _ConProxy:
+        """``DuckDBPyConnection`` has no ``__dict__`` (C extension type), so
+        a plain ``con.published = []`` fails; wrap it to attach the
+        publish-tracking list the tests assert on while forwarding
+        everything else (execute, executemany, ...) to the real connection."""
+
+        def __init__(self, real, published):
+            object.__setattr__(self, "_real", real)
+            object.__setattr__(self, "published", published)
+
+        def __getattr__(self, name):
+            return getattr(object.__getattribute__(self, "_real"), name)
+
+    try:
+        yield _ConProxy(c, published)
+    finally:
+        c.close()
+
+
+class TestElements:
+    def test_writes_rows_and_publishes(self, con, monkeypatch):
+        def fake_extract(store, con_, code, *, classify=None):
+            return Extraction(
+                code,
+                (
+                    ElementRow(
+                        code,
+                        2026,
+                        "activity",
+                        "consent",
+                        "",
+                        "Consent;",
+                        "K",
+                        1,
+                        2,
+                        "fr",
+                    ),
+                ),
+                (),
+            )
+
+        monkeypatch.setattr(pfs_cli, "extract_code", fake_extract)
+        res = runner.invoke(app, ["pfs", "elements", "--code", "g0556", "--no-llm"])
+        assert res.exit_code == 0, res.output
+        assert "G0556: 1 elements, 0 for review" in res.output
+        assert [r.value for r in read_elements(con, "G0556")] == ["consent"]
+        assert con.published == [True]
+
+    def test_dry_run_does_not_write(self, con, monkeypatch):
+        monkeypatch.setattr(
+            pfs_cli,
+            "extract_code",
+            lambda s, c, code, *, classify=None: Extraction(code, (), ()),
+        )
+        res = runner.invoke(
+            app, ["pfs", "elements", "--code", "G0556", "--no-llm", "--dry-run"]
+        )
+        assert res.exit_code == 0 and con.published == []
+
+    def test_all_payable_selects_art_status_in_newest_year(self, con, monkeypatch):
+        seen = []
+        monkeypatch.setattr(
+            pfs_cli,
+            "extract_code",
+            lambda s, c, code, *, classify=None: (
+                seen.append(code) or Extraction(code, (), ())
+            ),
+        )
+        res = runner.invoke(
+            app, ["pfs", "elements", "--all-payable", "--no-llm", "--dry-run"]
+        )
+        assert res.exit_code == 0 and seen == ["99490"]
+
+
+class TestLineage:
+    def test_prints_and_writes(self, con, monkeypatch):
+        ev = EventRow(
+            "G2058",
+            2021,
+            "replaced_by",
+            "",
+            "99439",
+            "YBM4IZUS",
+            1578,
+            84639,
+            "fr",
+            True,
+            "",
+        )
+        monkeypatch.setattr(pfs_cli, "lineage", lambda c, s, code: [ev])
+        res = runner.invoke(app, ["pfs", "lineage", "--code", "G2058", "--write"])
+        assert res.exit_code == 0, res.output
+        assert "2021  replaced_by" in res.output and "YBM4IZUS ¶1578" in res.output
+        assert read_events(con, "G2058")[0].to_codes == "99439"
+        assert con.published == [True]
+
+
+class TestFamilies:
+    def test_derives_from_tables_and_writes(self, con):
+        write_elements(
+            con,
+            "99439",
+            [
+                ElementRow(
+                    "99439", 2021, "relation", "addon-of", "99490", "", "K", 1, 1, "fr"
+                )
+            ],
+            [],
+        )
+        write_events(
+            con,
+            "G2058",
+            [
+                EventRow(
+                    "G2058", 2021, "replaced_by", "", "99439", "K", 2, 1, "fr", True, ""
+                )
+            ],
+        )
+        res = runner.invoke(app, ["pfs", "families", "--write"])
+        assert res.exit_code == 0, res.output
+        fams = read_families(con)
+        assert {r.code for r in fams if r.key == "CCM"} >= {"99490", "99439", "G2058"}
+        assert "CCM" in res.output
+
+
+class TestReview:
+    def test_lists_queue(self, con):
+        from pfs.codetables import ReviewRow
+
+        write_elements(
+            con, "G0556", [], [ReviewRow("G0556", "Odd line;", "", "", "K", 9)]
+        )
+        res = runner.invoke(app, ["pfs", "review"])
+        assert res.exit_code == 0 and "G0556" in res.output and "Odd line" in res.output