feat(pfs): lineage for every payable code — one inverted pass over fr_anchors, stack pfs lineage --all-payable (refs #691 #698)

src/cli/pfs.py

@@ -42,7 +42,7 @@ from pfs.cpt_load import ingest as cpt_ingest
 from pfs.extract import extract_code
 from pfs.families import FAMILIES, HAND_FAMILIES, derive_families, refresh_from
 from pfs.guidance import build as build_guidance
-from pfs.lineage import lineage
+from pfs.lineage import lineage, lineage_all
 from pfs.reaction import series as reaction_series
 
 log = logging.getLogger(__name__)
@@ -205,20 +205,83 @@ def _print_events(events: list[Any]) -> None:
         typer.echo(f"{e.year}  {e.kind:<18}{arrow:<16}{anchor}{flag}")
 
 
+def _print_lineage_all(by_code: dict[str, list[Any]]) -> None:
+    per_source: dict[str, int] = {}
+    with_events = 0
+    unanchored = 0
+    for events in by_code.values():
+        if events:
+            with_events += 1
+        for e in events:
+            per_source[e.source] = per_source.get(e.source, 0) + 1
+            if e.source == "rvu" and not e.anchored:
+                unanchored += 1
+    detail = ", ".join(f"{k}={v}" for k, v in sorted(per_source.items())) or "none"
+    typer.echo(
+        f"lineage: {len(by_code)} codes targeted, {with_events} with >=1 event, "
+        f"events per source: {detail}, unanchored rvu events: {unanchored}"
+    )
+
+
 @app.command("lineage")
 def lineage_cmd(
-    code: str = typer.Option(..., "--code"),
+    code: str = typer.Option("", "--code"),
+    all_payable: bool = typer.Option(
+        False,
+        "--all-payable",
+        help="Every A/R/T code in the newest RVU year, one inverted pass over fr_anchors.",
+    ),
+    limit: int = typer.Option(
+        0,
+        "--limit",
+        help="Only the first N target codes (sorted) — for smoke runs with --all-payable.",
+    ),
     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."""
+    """Timeline of a code: RVU-file diffs and FR paragraphs, cross-checked.
+
+    --all-payable computes lineage for every payable code in ONE pass over
+    fr_anchors instead of one query per code.
+    """
+    if not code and not all_payable:
+        raise typer.BadParameter("pass --code or --all-payable")
     store = _store()
+
+    if all_payable:
+        # Ruling A13: resolve targets and build every code's lineage
+        # against a plain DuckDB read (I4) — never the write lock a
+        # notebook may be holding (#508-#514). The batch below only ever
+        # calls write_events and publishes.
+        con = _read()
+        try:
+            targets = _codes_for(con, [], [], True)
+            if limit:
+                targets = targets[:limit]
+            by_code = lineage_all(con, store, targets)
+        finally:
+            con.close()
         if write:
             with _batch() as con:
                 ensure_tables(con)
+                for c in targets:
+                    write_events(con, c, by_code.get(c, []))
+            _publish()
+        _print_lineage_all(by_code)
+        return
+
+    if write:
+        # Same A13 discipline for the single-code path: build the events
+        # against a plain read, then open the batch only to write them.
+        con = _read()
+        try:
             events = lineage(con, store, code.upper())
+        finally:
+            con.close()
         _print_events(events)
+        with _batch() as con:
+            ensure_tables(con)
             write_events(con, code.upper(), events)
         _publish()
         return

src/pfs/lineage.py

@@ -23,7 +23,7 @@ from __future__ import annotations
 
 import dataclasses
 import re
-from typing import Any
+from typing import Any, Sequence
 
 from pfs.codetables import EventRow, cpt_years, is_missing_table_error
 from pfs.descriptors import rule_year_of
@@ -194,28 +194,39 @@ def _expand_ranges(text: str) -> str:
     return text
 
 
-def fr_events(store: Any, code: str) -> list[EventRow]:
+def _events_for(
+    code: str, rows: Sequence[Any], *, expanded: bool = False
+) -> list[EventRow]:
+    """The event-verb logic for *code* over already-fetched *rows*
+    (``item_key, p_id, page, text, title, date_published`` — a sqlite
+    ``Row`` mapping or a plain dict). ``expanded=True`` means ``text`` is
+    already range-expanded and FR-citation-stripped (``fr_events_bucketed``
+    computes that once per row, shared across every code it hits) — pass
+    ``expanded=False`` (the default) for raw ``fr_anchors.text``.
+
+    Presence and "other codes nearby" both go through ``find_codes`` (word
+    boundaries, FR-citation-safe) rather than a bare substring check, so
+    this agrees exactly with the ``find_codes(text) & targets`` hit test
+    ``fr_events_bucketed`` uses to decide which bucket a row lands in."""
     code = code.upper()
-    con = store._con()
-    rows = con.execute(
-        "SELECT a.item_key, a.p_id, a.page, a.text, i.title, i.date_published "
-        "FROM fr_anchors a JOIN items i ON i.key = a.item_key WHERE a.text LIKE ? OR a.text LIKE ?",
-        (f"%{code}%", f"%{code[:2]}___-{code[:2]}___%"),
-    ).fetchall()
     patterns = _patterns(code)
     seen: set[tuple[int, str, str, int]] = set()
     out: list[EventRow] = []
     for r in rows:
-        text = _expand_ranges(r["text"])
+        if expanded:
-        # An FR page citation's number ("91 FR 99490") is not a mention of
+            text = r["text"]
-        # the code — strip it before both the presence check and every
+        else:
-        # event pattern below (M1: a "crosswalk ... 91 FR 99490" sentence
+            # An FR page citation's number ("91 FR 99490") is not a
-        # used to read as a crosswalk event for 99490).
+            # mention of the code — strip it before both the presence
-        text = FR_CITE_RE.sub(" ", text)
+            # check and every event pattern below (M1: a "crosswalk ...
-        if code not in text.upper():
+            # 91 FR 99490" sentence used to read as a crosswalk event
+            # for 99490).
+            text = FR_CITE_RE.sub(" ", _expand_ranges(r["text"]))
+        codes_here = find_codes(text)
+        if code not in codes_here:
             continue
         year = rule_year_of(r["title"], r["date_published"])
-        others = tuple(c for c in find_codes(text) if c != code)
+        others = tuple(c for c in codes_here if c != code)
         for kind, pat, direction in patterns:
             m = pat.search(text)
             if not m:
@@ -265,6 +276,53 @@ def fr_events(store: Any, code: str) -> list[EventRow]:
     return sorted(out, key=lambda e: (e.year, FR_KINDS.index(e.kind), e.p_id))
 
 
+def fr_events(store: Any, code: str) -> list[EventRow]:
+    code = code.upper()
+    con = store._con()
+    rows = con.execute(
+        "SELECT a.item_key, a.p_id, a.page, a.text, i.title, i.date_published "
+        "FROM fr_anchors a JOIN items i ON i.key = a.item_key WHERE a.text LIKE ? OR a.text LIKE ?",
+        (f"%{code}%", f"%{code[:2]}___-{code[:2]}___%"),
+    ).fetchall()
+    return _events_for(code, rows)
+
+
+def fr_events_bucketed(store: Any, codes: Sequence[str]) -> dict[str, list[EventRow]]:
+    """``_events_for`` for every one of *codes* in ONE streaming pass over
+    ``fr_anchors`` instead of one SQL query (two LIKE prefilters) per
+    code — the whole point for an 8.7k-code universe over 193k
+    paragraphs. Each row's text is range-expanded and FR-citation-stripped
+    exactly once, then dropped into every target code's bucket that
+    ``find_codes`` finds it naming.
+
+    ``fr_events_bucketed(store, [c])[c] == fr_events(store, c)`` for any
+    single code ``c``: ``fr_events``'s LIKE prefilter is a superset of its
+    own presence check, and both presence checks now agree (``_events_for``
+    docstring), so the two paths select exactly the same rows for *c*."""
+    targets = {c.upper() for c in codes}
+    buckets: dict[str, list[dict[str, Any]]] = {c: [] for c in targets}
+    con = store._con()
+    for r in con.execute(
+        "SELECT a.item_key, a.p_id, a.page, a.text, i.title, i.date_published "
+        "FROM fr_anchors a JOIN items i ON i.key = a.item_key ORDER BY a.item_key, a.p_id"
+    ):
+        text = FR_CITE_RE.sub(" ", _expand_ranges(r["text"]))
+        hits = targets & set(find_codes(text))
+        if not hits:
+            continue
+        row = {
+            "item_key": r["item_key"],
+            "p_id": r["p_id"],
+            "page": r["page"],
+            "text": text,
+            "title": r["title"],
+            "date_published": r["date_published"],
+        }
+        for c in hits:
+            buckets[c].append(row)
+    return {c: _events_for(c, rows, expanded=True) for c, rows in buckets.items()}
+
+
 def _cpt_ev(code: str, year: int, kind: str, item_key: str, note: str) -> EventRow:
     return EventRow(code, year, kind, "", "", item_key, 0, 0, "cpt", True, note)
 
@@ -318,14 +376,51 @@ def cpt_events(con: Any, code: str) -> list[EventRow]:
     return []
 
 
-def lineage(con: Any, store: Any, code: str) -> list[EventRow]:
+def _merge(
-    fr = fr_events(store, code)
+    code: str,
-    cpt = cpt_events(con, code)
+    rvu: list[EventRow],
+    fr: list[EventRow],
+    cpt: list[EventRow],
+) -> list[EventRow]:
+    """Merge one code's rvu/fr/cpt events, marking each rvu event
+    ``anchored`` when an fr or cpt event for the same code lies within
+    ±1 rule year — shared by ``lineage`` (per-code) and ``lineage_all``
+    (bucketed) so both anchor identically."""
     anchor_years = {e.year for e in fr} | {e.year for e in cpt}
-    rvu = [
+    anchored_rvu = [
         dataclasses.replace(
             ev, anchored=any(abs(ev.year - y) <= 1 for y in anchor_years)
         )
-        for ev in rvu_events(con, code)
+        for ev in rvu
     ]
-    return sorted([*rvu, *fr, *cpt], key=lambda e: (e.year, e.source != "fr", e.kind))
+    return sorted(
+        [*anchored_rvu, *fr, *cpt], key=lambda e: (e.year, e.source != "fr", e.kind)
+    )
+
+
+def lineage(con: Any, store: Any, code: str) -> list[EventRow]:
+    fr = fr_events(store, code)
+    cpt = cpt_events(con, code)
+    rvu = rvu_events(con, code)
+    return _merge(code, rvu, fr, cpt)
+
+
+def lineage_all(
+    con: Any, store: Any, codes: Sequence[str]
+) -> dict[str, list[EventRow]]:
+    """``lineage`` for every one of *codes*, computing fr events with ONE
+    pass over ``fr_anchors`` (``fr_events_bucketed``) instead of one per
+    code; rvu/cpt events stay per-code (cheap replica reads) and are
+    merged with the same ±1-year anchoring ``lineage`` uses, via
+    ``_merge``."""
+    codes_u = [c.upper() for c in codes]
+    fr_by_code = fr_events_bucketed(store, codes_u)
+    return {
+        code: _merge(
+            code,
+            rvu_events(con, code),
+            fr_by_code.get(code, []),
+            cpt_events(con, code),
+        )
+        for code in codes_u
+    }

tests/cli/test_pfs_cli.py

@@ -270,6 +270,77 @@ class TestLineage:
         assert read_calls == [True]
         assert con.published == []
 
+    def test_requires_code_or_all_payable(self, con):
+        res = runner.invoke(app, ["pfs", "lineage"])
+        assert res.exit_code != 0
+
+    def test_all_payable_limit_write_calls_writer_once_per_code_after_build(
+        self, con, monkeypatch
+    ):
+        # Extra A/R/T, newest-year codes so --limit has something to
+        # trim: sorted, the newest-year A/R/T targets are
+        # 99490 < A0001 < B0002 ('9' sorts before letters); 99999 is
+        # status B (excluded) and G2058 is a stale year (excluded).
+        con.execute(
+            "INSERT INTO pfs.rvu VALUES (?,?,?,?,?,?)",
+            ["A0001", None, "extra a", "A", 1.0, 2026],
+        )
+        con.execute(
+            "INSERT INTO pfs.rvu VALUES (?,?,?,?,?,?)",
+            ["B0002", None, "extra b", "R", 1.0, 2026],
+        )
+        order: list[tuple] = []
+
+        def fake_lineage_all(con_, store_, codes):
+            order.append(("lineage_all", tuple(codes)))
+            return {c: [] for c in codes}
+
+        monkeypatch.setattr(pfs_cli, "lineage_all", fake_lineage_all)
+
+        written: list[str] = []
+
+        def fake_write_events(con_, code, rows):
+            order.append(("write", code))
+            written.append(code)
+            return 0
+
+        monkeypatch.setattr(pfs_cli, "write_events", fake_write_events)
+
+        real_batch = pfs_cli._batch
+
+        def recording_batch():
+            order.append(("batch",))
+            return real_batch()
+
+        monkeypatch.setattr(pfs_cli, "_batch", recording_batch)
+
+        res = runner.invoke(
+            app, ["pfs", "lineage", "--all-payable", "--limit", "2", "--write"]
+        )
+        assert res.exit_code == 0, res.output
+        assert order[0] == ("lineage_all", ("99490", "A0001"))
+        assert order[1] == ("batch",)
+        assert order[2:] == [("write", "99490"), ("write", "A0001")]
+        assert written == ["99490", "A0001"]
+        assert con.published == [True]
+
+    def test_all_payable_without_write_never_opens_the_batch(self, con, monkeypatch):
+        monkeypatch.setattr(
+            pfs_cli, "lineage_all", lambda c, s, codes: {c: [] for c in codes}
+        )
+
+        def fail_batch():
+            raise AssertionError(
+                "lineage --all-payable without --write must not open a RW "
+                "duckdb_batch connection"
+            )
+
+        monkeypatch.setattr(pfs_cli, "_batch", fail_batch)
+        res = runner.invoke(app, ["pfs", "lineage", "--all-payable"])
+        assert res.exit_code == 0, res.output
+        assert con.published == []
+        assert "lineage: 1 codes targeted" in res.output
+
 
 class TestFamilies:
     def test_derives_from_tables_and_writes(self, con):

tests/pfs/test_lineage.py

@@ -12,7 +12,9 @@ from pfs.lineage import (
     _expand_ranges,
     cpt_events,
     fr_events,
+    fr_events_bucketed,
     lineage,
+    lineage_all,
     rvu_events,
     rvu_year_span,
 )
@@ -363,3 +365,73 @@ class TestCrossCheck:
     def test_sorted_by_year_then_source(self, con, store):
         ev = lineage(con, store, "99439")
         assert [e.year for e in ev] == sorted(e.year for e in ev)
+
+
+class TestFrEventsBucketed:
+    # The four target codes from `store`'s three rule items: a plain
+    # mention (99490), a replaced_by/crosswalk pair (G2058 -> 99439) and
+    # a range-form deletion (99441, from "99441-99443").
+    _CODES = ("99490", "G2058", "99439", "99441")
+
+    def test_matches_fr_events_per_code(self, store):
+        bucketed = fr_events_bucketed(store, self._CODES)
+        for c in self._CODES:
+            assert bucketed[c] == fr_events(store, c), c
+
+    def test_range_form_middle_member_matches(self, store):
+        # 90868 is a middle member of a "90867-90869" range and appears
+        # nowhere else in the paragraph literally — both paths must find
+        # it via range expansion, not a literal LIKE hit.
+        bucketed = fr_events_bucketed(store, ["90868"])
+        assert bucketed["90868"] == fr_events(store, "90868")
+        assert any(e.kind == "deleted" for e in bucketed["90868"])
+
+    def test_fr_citation_page_number_is_not_a_mention(self, store):
+        # M1, bucketed path: a page citation ("91 FR 99490") must not
+        # make 99490 register a hit at all, not just skip the crosswalk
+        # pattern — matching fr_events's own guard.
+        store.con.execute(
+            "INSERT INTO fr_anchors VALUES (?,?,?,?,?)",
+            (
+                "DE2VH9PD",
+                1500,
+                67999,
+                1500,
+                "We note the crosswalk methodology discussed at 91 FR 99490 for related codes.",
+            ),
+        )
+        bucketed = fr_events_bucketed(store, ["99490"])
+        assert bucketed["99490"] == fr_events(store, "99490")
+        assert not any(e.kind == "crosswalk" for e in bucketed["99490"])
+
+    def test_unmentioned_code_gets_empty_bucket(self, store):
+        assert fr_events_bucketed(store, ["00000"]) == {"00000": []}
+
+
+class TestLineageAll:
+    def test_matches_lineage_per_code(self, con, store):
+        codes = ["99487", "G2058", "99439"]
+        by_code = lineage_all(con, store, codes)
+        for c in codes:
+            assert by_code[c] == lineage(con, store, c), c
+
+    def test_rvu_event_anchored_by_cpt_changed_within_one_year(self, con, store):
+        _cpt_section(con, 2024, "GQGTPGYV")
+        _cpt_reference(con, 2024, "GQGTPGYV", "99487", [2016])
+        by_code = lineage_all(con, store, ["99487"])
+        by = {(e.kind, e.year): e for e in by_code["99487"]}
+        assert by[("status_change", 2017)].anchored is True
+        assert ("cpt_changed", 2016) in by
+
+    def test_rvu_events_anchored_by_nearby_fr_event(self, con, store):
+        by_code = lineage_all(con, store, ["G2058", "99487"])
+        by_kind = {e.kind: e for e in by_code["G2058"]}
+        assert by_kind["appeared"].year == 2020 and by_kind["appeared"].anchored is True
+        assert (
+            by_kind["disappeared"].year == 2021
+            and by_kind["disappeared"].anchored is True
+        )
+        assert all(not e.anchored for e in by_code["99487"] if e.source == "rvu")
+
+    def test_unknown_code_yields_empty_list(self, con, store):
+        assert lineage_all(con, store, ["99999"]) == {"99999": []}