fix(pfs): reaction FR pairs span continuation paragraphs, prefiltered scan; classify before the write lock (refs #690)

src/cli/pfs.py

@@ -96,6 +96,10 @@ def _classifier() -> Any:
 
 
 def _engine() -> Any:
+    # llm.index/llm.evidence expose no public "give me a bare SQLAlchemy
+    # engine over the pgvector DB" helper — cli/bib.py and cli/llm.py
+    # both already reach into llm.index._engine for the same reason, so
+    # this is the established reuse, not a new private import.
     from llm import config as llm_config
     from llm.index import _engine as pgvector_engine
 
@@ -445,9 +449,13 @@ def _print_reaction(rows: list[Any]) -> None:
 
 def _reaction_targets(
     con: Any, family: list[str], code: list[str]
-) -> list[tuple[str, tuple[str, ...]]]:
+) -> list[tuple[str, tuple[str, ...], str | None]]:
+    """(key, codes, display name) per target — *name* is the registered
+    family's own name for a ``--family`` target, ``None`` for a
+    ``--code`` pseudo-family (``pfs.reaction.series`` falls back to the
+    key itself, which is already the right label for a bare code)."""
     refresh_from(con)
-    targets: list[tuple[str, tuple[str, ...]]] = []
+    targets: list[tuple[str, tuple[str, ...], str | None]] = []
     for f in family:
         key = f.upper()
         fam = FAMILIES.get(key)
@@ -455,10 +463,10 @@ def _reaction_targets(
             raise typer.BadParameter(
                 f"unknown family {key!r}; known: {', '.join(HAND_FAMILIES)}"
             )
-        targets.append((key, fam.codes))
+        targets.append((key, fam.codes, fam.name))
     for c in code:
         key = c.upper()
-        targets.append((key, (key,)))
+        targets.append((key, (key,), None))
     return targets
 
 
@@ -491,53 +499,44 @@ def reaction(
         raise typer.BadParameter("pass --family or --code (repeatable)")
     store = _store()
 
-    if write:
+    # Resolving family keys is a plain DuckDB read (I4) — never the
-        # Mirrors `elements --write` (I9): the stance classifier — the
+    # write lock a notebook may be holding (#508-#514) — and it's the
-        # only genuinely slow part, ~1-2s/sample — runs inside the batch
+    # only DuckDB touch this command needs before writing.
-        # here too, the same tradeoff that command already makes.
-        with _batch() as con:
-            ensure_tables(con)
-            targets = _reaction_targets(con, family, code)
-            classify = _classifier() if stance_sample > 0 else None
-            engine = _engine()
-            rows: list[Any] = []
-            for key, codes in targets:
-                fam_rows = reaction_series(
-                    engine,
-                    store,
-                    key,
-                    codes,
-                    classify=classify,
-                    stance_sample_n=stance_sample,
-                )
-                write_reaction(con, key, fam_rows)
-                rows.extend(fam_rows)
-        _publish()
-        _print_reaction(rows)
-        return
-
-    # No --write: resolving family keys is the only thing that touches
-    # DuckDB, and it's a plain read (I4) — never the write lock a
-    # notebook may be holding (#508-#514).
     con = _read()
     try:
         targets = _reaction_targets(con, family, code)
     finally:
         con.close()
+
+    # Ruling A13: the classifier and every family's series (docket
+    # counts, the stance sample — ~1-2s/sample — and FR pairs) all run
+    # against pgvector/bib, not DuckDB, so they run BEFORE any DuckDB
+    # batch is opened. The batch below only ever calls write_reaction
+    # and publishes — it never holds the single-writer lock (#508-#514)
+    # while the slow work happens.
     classify = _classifier() if stance_sample > 0 else None
     engine = _engine()
-    rows = []
+    by_family: dict[str, list[Any]] = {}
-    for key, codes in targets:
+    rows: list[Any] = []
-        rows.extend(
+    for key, codes, name in targets:
-            reaction_series(
+        fam_rows = reaction_series(
-                engine,
+            engine,
-                store,
+            store,
-                key,
+            key,
-                codes,
+            codes,
-                classify=classify,
+            classify=classify,
-                stance_sample_n=stance_sample,
+            stance_sample_n=stance_sample,
-            )
+            name=name,
         )
+        by_family[key] = fam_rows
+        rows.extend(fam_rows)
+
+    if write:
+        with _batch() as con:
+            ensure_tables(con)
+            for key, fam_rows in by_family.items():
+                write_reaction(con, key, fam_rows)
+        _publish()
     _print_reaction(rows)
 
 

src/pfs/reaction.py

@@ -168,44 +168,68 @@ _RESPONSE_RE = re.compile(r"^\s*Response\s*:", re.I)
 
 def fr_pairs(store: Any, codes: Sequence[str]) -> list[tuple[str, int, int, int]]:
     """``(item_key, rule_year, n_comment_paras, n_response_paras)`` per
-    rule item: a ``Comment:`` paragraph immediately followed (next
+    rule item — the pre-2017 proxy for public reaction, since
-    ``p_id`` in that item) by a ``Response:`` paragraph, where either
+    regulations.gov comment text isn't indexed that far back.
-    paragraph names one of *codes* — the pre-2017 proxy for public
+
-    reaction, since regulations.gov comment text isn't indexed that far
+    Ruling A12: a ``Comment:`` paragraph pairs with the *next*
-    back. Rules with zero qualifying pairs are omitted."""
+    ``Response:`` paragraph in the same item that comes before the next
+    ``Comment:`` paragraph — intervening continuation paragraphs (the
+    comment's own prose, wrapped across more than one ``fr_anchors``
+    row) are allowed between them, and the code match runs over the
+    whole span (comment + continuations + response), not just the
+    comment/response paragraphs themselves. A ``Comment:`` with no
+    ``Response:`` before the next ``Comment:`` (or the end of the item)
+    is unpaired and not counted. Rules with zero qualifying pairs are
+    omitted.
+
+    One SQL per family, scoped to ``Comment:``/``Response:`` openers
+    plus any paragraph naming one of *codes* (the same LIKE-prefilter
+    idiom ``pfs.guidance.harvest`` uses) — safe under the pairing rule
+    above: a dropped paragraph is prose that names no code and opens
+    neither a comment nor a response, so it can neither create nor hide
+    a pair.
+    """
     if not codes:
         return []
     con = store._con()  # noqa: SLF001 — same pattern as pfs.descriptors/pfs.guidance
+    code_likes = " OR ".join("a.text LIKE ?" for _ in codes)
+    params = [f"%{c.upper()}%" for c in codes]
     rows = con.execute(
         "SELECT a.item_key, a.p_id, 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"
+        f"WHERE a.text LIKE 'Comment:%' OR a.text LIKE 'Response:%' OR ({code_likes}) "
+        "ORDER BY a.item_key, a.p_id",
+        params,
     ).fetchall()
 
     pattern = _code_pattern(codes)
-    by_item: dict[str, list[tuple[int, str]]] = {}
+    by_item: dict[str, list[str]] = {}
     meta: dict[str, tuple[str, str]] = {}
-    for item_key, p_id, para_text, title, date_published in rows:
+    for item_key, _p_id, para_text, title, date_published in rows:
-        by_item.setdefault(item_key, []).append((p_id, para_text))
+        by_item.setdefault(item_key, []).append(para_text or "")
         meta[item_key] = (title, date_published)
 
     out: list[tuple[str, int, int, int]] = []
     for item_key, paras in by_item.items():
         n_comment = 0
         n_response = 0
-        for i, (_p_id, para_text) in enumerate(paras):
+        span: list[str] | None = None  # None: no comment currently open
-            if not _COMMENT_RE.match(para_text or ""):
+        for para_text in paras:
+            if _COMMENT_RE.match(para_text):
+                # A new Comment: abandons any still-open, unresolved one
+                # (Ruling A12: unpaired, never counted).
+                span = [para_text]
                 continue
-            if i + 1 >= len(paras):
+            if _RESPONSE_RE.match(para_text):
+                if span is not None:
+                    span.append(para_text)
+                    if _codes_in(" ".join(span), pattern):
+                        n_comment += 1
+                        n_response += 1
+                    span = None
                 continue
-            _next_p_id, next_text = paras[i + 1]
+            if span is not None:
-            if not _RESPONSE_RE.match(next_text or ""):
+                span.append(para_text)  # continuation paragraph
-                continue
-            combined = f"{para_text} {next_text}"
-            if not _codes_in(combined, pattern):
-                continue
-            n_comment += 1
-            n_response += 1
         if n_comment:
             title, date_published = meta[item_key]
             out.append(
@@ -222,14 +246,20 @@ def series(
     *,
     classify: Callable[[str, Sequence[str]], str | None] | None = None,
     stance_sample_n: int = 0,
+    name: str | None = None,
 ) -> list[Any]:
     """Every ``ReactionRow`` for one family (or ``--code`` pseudo-family):
     a docket row per docket with any comment chunks, then an fr-pairs row
     per rule with qualifying Comment/Response pairs. Stance columns stay
     zero (and ``sample_json`` ``"[]"``) unless *classify* is given and
-    *stance_sample_n* > 0."""
+    *stance_sample_n* > 0. *name* is the family's display name for the
+    stance prompt (``FAMILIES[key].name`` for a registered family); it
+    defaults to *family* itself, which is already the right label for a
+    ``--code`` pseudo-family (there is no separate display name for a
+    single code)."""
     from pfs.codetables import ReactionRow
 
+    label = name or family
     docket_years = {d.id: d.comment_end_date[:4] for d in store.dockets()}
 
     rows: list[ReactionRow] = []
@@ -241,7 +271,7 @@ def series(
         if classify is not None and stance_sample_n > 0:
             sample = stance_sample(engine, family, codes, docket, n=stance_sample_n)
             stances = [
-                (item_key, _stance_of(_prompt_text(family, codes, chunk), classify))
+                (item_key, _stance_of(_prompt_text(label, codes, chunk), classify))
                 for item_key, chunk in sample
             ]
             stance_counts = Counter(s for _, s in stances)

tests/cli/test_pfs_cli.py

@@ -69,7 +69,13 @@ def con(monkeypatch, restore_families):
             return False
 
     monkeypatch.setattr(pfs_cli, "_batch", lambda: _Batch())
-    monkeypatch.setattr(pfs_cli, "_read", lambda: c)
+    # A DuckDB cursor, not `c` itself: `stack pfs reaction` (unlike the
+    # other commands here) calls `_read()` and then, on `--write`,
+    # `_batch()` in the same invocation — closing a `.close()`d `c`
+    # would break the batch. A cursor shares `c`'s catalog/data and
+    # closes independently of it, matching how `_read()`/`_batch()` are
+    # genuinely separate connections in production.
+    monkeypatch.setattr(pfs_cli, "_read", lambda: c.cursor())
     monkeypatch.setattr(pfs_cli, "_store", lambda: object())
     published = []
     monkeypatch.setattr(pfs_cli, "_publish", lambda: published.append(True))
@@ -633,7 +639,14 @@ class TestReaction:
         seen = {}
 
         def fake_series(
-            engine, store_, family, codes, *, classify=None, stance_sample_n=0
+            engine,
+            store_,
+            family,
+            codes,
+            *,
+            classify=None,
+            stance_sample_n=0,
+            name=None,
         ):
             seen["engine"] = engine
             seen["store"] = store_
@@ -641,6 +654,7 @@ class TestReaction:
             seen["codes"] = codes
             seen["classify"] = classify
             seen["n"] = stance_sample_n
+            seen["name"] = name
             return [self._row(family=family)]
 
         monkeypatch.setattr(pfs_cli, "reaction_series", fake_series)
@@ -651,20 +665,31 @@ class TestReaction:
         assert seen["engine"] == "ENGINE"
         assert seen["classify"] is None  # no --stance-sample: no LLM at all
         assert seen["n"] == 0
+        assert seen["name"] == pfs_cli.HAND_FAMILIES["CCM"].name
         assert set(seen["codes"]) == set(pfs_cli.HAND_FAMILIES["CCM"].codes)
         got = read_reaction(con, "CCM")
         assert [r.period for r in got] == ["CMS-2023-0121"]
         assert con.published == [True]
         assert "CCM  CMS-2023-0121  (2023)  5/20 commenters" in res.output
 
-    def test_code_makes_a_single_code_pseudo_family(self, con, monkeypatch):
+    def test_code_makes_a_single_code_pseudo_family_with_no_display_name(
+        self, con, monkeypatch
+    ):
         seen = {}
 
         def fake_series(
-            engine, store_, family, codes, *, classify=None, stance_sample_n=0
+            engine,
+            store_,
+            family,
+            codes,
+            *,
+            classify=None,
+            stance_sample_n=0,
+            name=None,
         ):
             seen["family"] = family
             seen["codes"] = codes
+            seen["name"] = name
             return [self._row(family=family)]
 
         monkeypatch.setattr(pfs_cli, "reaction_series", fake_series)
@@ -673,6 +698,7 @@ class TestReaction:
         assert res.exit_code == 0, res.output
         assert seen["family"] == "G2211"
         assert seen["codes"] == ("G2211",)
+        assert seen["name"] is None
         assert [r.family for r in read_reaction(con, "G2211")] == ["G2211"]
 
     def test_stance_sample_builds_the_classifier(self, con, monkeypatch):
@@ -680,7 +706,7 @@ class TestReaction:
         monkeypatch.setattr(
             pfs_cli,
             "reaction_series",
-            lambda engine, store_, family, codes, *, classify=None, stance_sample_n=0: (
+            lambda engine, store_, family, codes, *, classify=None, stance_sample_n=0, name=None: (  # noqa: E501
                 seen.update(classify=classify, n=stance_sample_n) or []
             ),
         )
@@ -693,11 +719,54 @@ class TestReaction:
         assert seen["classify"] == "CLASSIFIER"
         assert seen["n"] == 5
 
+    def test_write_computes_classifier_and_series_before_opening_the_batch(
+        self, con, monkeypatch
+    ):
+        # Ruling A13: the classifier and every family's series must be
+        # fully computed BEFORE the DuckDB write lock is taken — the
+        # batch block only writes and publishes.
+        order: list[str] = []
+
+        def fake_classifier():
+            order.append("classifier")
+            return "CLASSIFIER"
+
+        def fake_series(
+            engine,
+            store_,
+            family,
+            codes,
+            *,
+            classify=None,
+            stance_sample_n=0,
+            name=None,
+        ):
+            order.append("series")
+            assert classify == "CLASSIFIER"  # already built by the time series runs
+            return [self._row(family=family)]
+
+        real_batch = pfs_cli._batch
+
+        def recording_batch():
+            order.append("batch")
+            return real_batch()
+
+        monkeypatch.setattr(pfs_cli, "_classifier", fake_classifier)
+        monkeypatch.setattr(pfs_cli, "reaction_series", fake_series)
+        monkeypatch.setattr(pfs_cli, "_engine", lambda: "ENGINE")
+        monkeypatch.setattr(pfs_cli, "_batch", recording_batch)
+        res = runner.invoke(
+            app,
+            ["pfs", "reaction", "--family", "CCM", "--stance-sample", "5", "--write"],
+        )
+        assert res.exit_code == 0, res.output
+        assert order == ["classifier", "series", "batch"]
+
     def test_without_write_never_opens_a_batch_connection(self, con, monkeypatch):
         monkeypatch.setattr(
             pfs_cli,
             "reaction_series",
-            lambda engine, store_, family, codes, *, classify=None, stance_sample_n=0: [
+            lambda engine, store_, family, codes, *, classify=None, stance_sample_n=0, name=None: [  # noqa: E501
                 self._row(family=family)
             ],
         )

tests/pfs/test_reaction.py

@@ -141,6 +141,46 @@ def store_with_pairs():
     s.close()
 
 
+@pytest.fixture
+def store_with_continuation_pair():
+    """Ruling A12: a Comment: paragraph with no code of its own, a
+    continuation paragraph (no Comment:/Response: prefix) naming a code,
+    then the Response: — must still count as one pair."""
+    s = Store(":memory:")
+    item_key = s.create(Rule(title="CY2016 PFS final rule"))
+    con = s._con()  # noqa: SLF001
+    con.executemany(
+        "INSERT INTO fr_anchors (item_key, p_id, page, ordinal, text) VALUES (?,?,?,?,?)",
+        [
+            (
+                item_key,
+                20,
+                200,
+                1,
+                "Comment: Commenters raised concerns about staff time requirements.",
+            ),
+            (
+                item_key,
+                21,
+                200,
+                2,
+                "Providers noted that 99490 requires clinical staff supervision "
+                "that is hard to bill separately.",
+            ),
+            (
+                item_key,
+                22,
+                200,
+                3,
+                "Response: We agree and are clarifying supervision requirements.",
+            ),
+        ],
+    )
+    con.commit()
+    yield s, item_key
+    s.close()
+
+
 class TestFrPairs:
     def test_counts_only_pairs_naming_a_code(self, store_with_pairs):
         s, item_key = store_with_pairs
@@ -163,6 +203,40 @@ class TestFrPairs:
         out = fr_pairs(s, ("99490",))
         assert out == [(item_key, 2015, 1, 1)]
 
+    def test_code_named_only_in_a_continuation_paragraph_still_pairs(
+        self, store_with_continuation_pair
+    ):
+        # Ruling A12: intervening continuation paragraphs are allowed
+        # between Comment: and Response:, and the code match runs over
+        # the whole span — neither the Comment: nor the Response:
+        # paragraph names 99490 here, only the continuation between them.
+        s, item_key = store_with_continuation_pair
+        out = fr_pairs(s, ("99490",))
+        assert out == [(item_key, 2016, 1, 1)]
+
+    def test_new_comment_abandons_an_unresolved_open_one(self):
+        # A Comment: immediately followed by another Comment: (no
+        # Response: in between) leaves the first one unpaired; the
+        # second one, properly followed by a Response:, still counts.
+        s = Store(":memory:")
+        item_key = s.create(Rule(title="CY2017 PFS final rule"))
+        con = s._con()  # noqa: SLF001
+        con.executemany(
+            "INSERT INTO fr_anchors (item_key, p_id, page, ordinal, text) "
+            "VALUES (?,?,?,?,?)",
+            [
+                (item_key, 1, 1, 1, "Comment: A commenter mentioned 99490 in passing."),
+                (item_key, 2, 1, 2, "Comment: A second, unrelated 99490 comment."),
+                (item_key, 3, 1, 3, "Response: We are finalizing 99490 as proposed."),
+            ],
+        )
+        con.commit()
+        try:
+            out = fr_pairs(s, ("99490",))
+        finally:
+            s.close()
+        assert out == [(item_key, 2017, 1, 1)]
+
 
 # ── series ──────────────────────────────────────────────────────────
 
@@ -221,6 +295,31 @@ class TestSeries:
         assert {s_["item_key"] for s_ in sample} == {"C1", "C2"}
         assert {s_["stance"] for s_ in sample} == {"support", "oppose"}
 
+    def test_name_overrides_family_key_in_the_stance_prompt(self, store):
+        s, _item_key = store
+        engine, _conn = _multi_engine(
+            docket_rows=[("CMS-2023-0121", 5, 20)],
+            stance_rows=[("C1", "a comment")],
+        )
+        seen_prompts = []
+
+        def fake_classify(text, choices):
+            seen_prompts.append(text)
+            return "support"
+
+        series(
+            engine,
+            s,
+            "CCM",
+            ("99490",),
+            classify=fake_classify,
+            stance_sample_n=1,
+            name="Chronic Care Management",
+        )
+        assert len(seen_prompts) == 1
+        assert "Chronic Care Management" in seen_prompts[0]
+        assert "CCM" not in seen_prompts[0]
+
     def test_zero_stance_sample_skips_classification(self, store):
         s, _item_key = store
         engine, conn = _multi_engine(docket_rows=[("CMS-2023-0121", 5, 20)])