diff --git a/src/cli/pfs.py b/src/cli/pfs.py
index 534dc4c..f0716f5 100644
--- a/src/cli/pfs.py
+++ b/src/cli/pfs.py
@@ -677,16 +677,29 @@ def _stored(read: Any, con: Any, family: str) -> list[Any]:
raise
-def _overlay_stances(rows: list[Any], stances: list[Any]) -> list[Any]:
+def _overlay_stances(
+ rows: list[Any], stances: list[Any], stored: list[Any], commenters: Counter
+) -> list[Any]:
"""Put the classified counts of each docket's stored stances on its row,
- so a reaction rerun never zeroes them."""
+ so a reaction rerun never zeroes them. The event order is the one the
+ docket's last ``stance`` run stored (*stored* are the family's stored
+ reaction rows); *commenters* counts ``pfs.commenter`` rows per docket (R62)."""
+ old = {r.period: r for r in stored}
out = []
for r in rows:
mine = [s for s in stances if s.docket == r.period]
- out.append(rollup(mine, base=r) if r.period_kind == "docket" and mine else r)
+ if r.period_kind != "docket" or not mine:
+ out.append(r)
+ continue
+ order = list(json.loads(old[r.period].events_json)) if r.period in old else []
+ out.append(rollup(mine, base=r, order=order, n_commenters=commenters[r.period]))
return out
+def _commenters_of(con: Any, family: str) -> list[Any]:
+ return read_commenters(con, family=family)
+
+
@app.command()
def reaction(
family: list[str] = typer.Option(
@@ -723,6 +736,12 @@ def reaction(
try:
targets = _reaction_targets(con, family, code)
stored = {key: _stored(read_stances, con, key) for key, _, _ in targets}
+ old_rows = {key: _stored(read_reaction, con, key) for key in stored}
+ n_comm = {
+ key: Counter(c.docket for c in _stored(_commenters_of, con, key))
+ for key, v in stored.items()
+ if v
+ }
finally:
con.close()
if stance_sample > 0:
@@ -753,7 +772,9 @@ def reaction(
stance_sample_n=stance_sample,
name=name,
)
- fam_rows = _overlay_stances(fam_rows, stored[key])
+ fam_rows = _overlay_stances(
+ fam_rows, stored[key], old_rows[key], n_comm.get(key, Counter())
+ )
by_family[key] = fam_rows
rows.extend(fam_rows)
@@ -1346,6 +1367,9 @@ def stance(
dry_run: bool = typer.Option(
False, "--dry-run", help="Count commenters and events; no model."
),
+ no_llm: bool = typer.Option(
+ False, "--no-llm", help="Make no model calls; the finished ledger is replayed."
+ ),
write: bool = typer.Option(False, "--write"),
) -> None:
"""Each commenter's stance (support / modify / oppose / unclear) on each
@@ -1362,7 +1386,7 @@ def stance(
year = docket_rule_year(found[0].comment_end_date)
con = _read()
try:
- events = docket_events(con, fam, year)
+ events = docket_events(con, fam, year, store)
commenter_rows = [
c for c in read_commenters(con, family=fam.key) if c.docket == docket
]
@@ -1387,7 +1411,7 @@ def stance(
f"text, {len(events)} events"
)
for e in events:
- typer.echo(f" {e.event_key}")
+ typer.echo(f" {e.event_key}: {e.text}")
if dry_run:
typer.echo("no model calls made")
return
@@ -1395,23 +1419,25 @@ def stance(
commenter_rows,
texts,
events,
- _asker(),
+ None if no_llm else _asker(),
RunLog(ledger_path("stance", fam.key)),
family=fam.key,
docket=docket,
+ model=_model_name() if no_llm else None,
limit=limit,
retry_failed=retry_failed,
progress=lambda unit, status: typer.echo(f"{unit} {status}"),
)
blank = ReactionRow(fam.key, docket, "docket", year, 0, 0, 0, 0, 0, 0, "[]")
- rolled = rollup(rows, base=blank)
+ order = [e.event_key for e in events]
+ rolled = rollup(rows, base=blank, order=order, n_commenters=len(commenter_rows))
for key, n in json.loads(rolled.events_json).items():
typer.echo(
f"{key}: support {n['support']}, modify {n['modify']}, "
f"oppose {n['oppose']}, unclear {n['unclear']}"
)
typer.echo(
- f"commenters classified: {len({r.item_key for r in rows})}; "
+ f"commenters classified: {rolled.n_classified} of {len(commenter_rows)}; "
f"support {rolled.stance_support}, modify {rolled.stance_modify}, "
f"oppose {rolled.stance_oppose}, unclear {rolled.stance_unclear}"
)
@@ -1431,7 +1457,7 @@ def stance(
(r for r in series_rows if r.period_kind == "docket" and r.period == docket),
blank,
)
- new = rollup(rows, base=base)
+ new = rollup(rows, base=base, order=order, n_commenters=len(commenter_rows))
keep = [
r
for r in (stored or series_rows)
diff --git a/src/pfs/codetables.py b/src/pfs/codetables.py
index 43664e1..deca888 100644
--- a/src/pfs/codetables.py
+++ b/src/pfs/codetables.py
@@ -169,6 +169,12 @@ class ReactionRow:
#: JSON ``{event_key: {"support": n, "modify": n, "oppose": n, "unclear": n}}``
#: from ``pfs.comment_stance`` (task 9); ``"{}"`` when none was classified
events_json: str = "{}"
+ #: R62: distinct commenters with at least one firm (non-``unclear``)
+ #: position; ``None`` = this docket was never classified
+ n_classified: int | None = None
+ #: R62: ``pfs.commenter`` rows for (family, docket); ``None`` = never
+ #: classified
+ n_commenters: int | None = None
@dataclass(frozen=True)
@@ -467,7 +473,7 @@ CREATE TABLE IF NOT EXISTS pfs.code_reaction (
family VARCHAR, period VARCHAR, period_kind VARCHAR, year INTEGER,
n_items INTEGER, n_total INTEGER, stance_support INTEGER, stance_oppose INTEGER,
stance_modify INTEGER, stance_unclear INTEGER, sample_json VARCHAR,
- events_json VARCHAR);
+ events_json VARCHAR, n_classified INTEGER, n_commenters INTEGER);
CREATE TABLE IF NOT EXISTS pfs.cpt_section (
edition_year INTEGER, item_key VARCHAR, sec_id VARCHAR, level INTEGER, title VARCHAR,
path VARCHAR[], path_key VARCHAR, code_lo VARCHAR, code_hi VARCHAR, guideline VARCHAR);
@@ -568,6 +574,12 @@ def ensure_tables(con: Any) -> None:
con.execute(
"ALTER TABLE pfs.code_reaction ADD COLUMN IF NOT EXISTS events_json VARCHAR"
)
+ con.execute(
+ "ALTER TABLE pfs.code_reaction ADD COLUMN IF NOT EXISTS n_classified INTEGER"
+ )
+ con.execute(
+ "ALTER TABLE pfs.code_reaction ADD COLUMN IF NOT EXISTS n_commenters INTEGER"
+ )
con.execute("UPDATE pfs.commenter SET family = 'HOME-VISITS' WHERE family IS NULL")
@@ -717,8 +729,10 @@ def read_reaction(con: Any, family: str) -> list[ReactionRow]:
"ORDER BY period_kind, year, period",
[family],
).fetchall()
- # rows written before ``events_json`` existed read back NULL for it
- return [ReactionRow(*r[:-1], r[-1] or "{}") for r in rows]
+ # a replica published before ``events_json``/``n_classified``/
+ # ``n_commenters`` existed has shorter rows; NULL events read as ``{}``
+ padded = [(*r, None, None, None)[:14] for r in rows]
+ return [ReactionRow(*r[:11], r[11] or "{}", r[12], r[13]) for r in padded]
_CPT_TABLES = (
diff --git a/src/pfs/curated.py b/src/pfs/curated.py
index 9176123..73a9fd3 100644
--- a/src/pfs/curated.py
+++ b/src/pfs/curated.py
@@ -42,6 +42,11 @@ class CuratedLink:
p_id: int
page: int
rvu_year: int
+ #: placeholder codes the proposed rule used for this link ("GXXX3" for
+ #: G0318), cited to the proposed-rule paragraph that introduces them
+ aliases: tuple[str, ...] = ()
+ alias_item_key: str = ""
+ alias_p_id: int = 0
@dataclass(frozen=True)
@@ -87,6 +92,14 @@ def _cite(entry: Mapping[str, Any], where: str) -> tuple[str, int, int, int]:
raise ValueError(f"{where}: ambiguous cite {sorted(keys)}")
+def _alias_cite(entry: Mapping[str, Any], where: str) -> tuple[str, int, int, int]:
+ """The ``alias_cite`` of a link with aliases: a paragraph, nothing else."""
+ cite = entry.get("alias_cite")
+ if not isinstance(cite, dict) or set(cite) != {"item_key", "p_id"}:
+ raise ValueError(f"{where}: aliases need alias_cite {{item_key, p_id}}")
+ return _cite({"cite": cite}, where)
+
+
def _year(entry: Mapping[str, Any], field: str, where: str) -> int | None:
v = entry.get(field)
return None if v is None else _int(v, where, field)
@@ -138,6 +151,14 @@ def _parse(key: str, raw: Any) -> CuratedFamily:
kind = ln.get("kind")
if kind not in KINDS:
raise ValueError(f"{where}: unknown kind {kind!r}")
+ aliases = ln.get("aliases") or []
+ if not isinstance(aliases, list) or not all(
+ isinstance(a, str) for a in aliases
+ ):
+ raise ValueError(f"{where}: aliases must be a list of strings")
+ alias_item, alias_p = "", 0
+ if aliases:
+ alias_item, alias_p, _, _ = _alias_cite(ln, where)
links.append(
CuratedLink(
kind,
@@ -145,6 +166,9 @@ def _parse(key: str, raw: Any) -> CuratedFamily:
_year(ln, "since", where),
_year(ln, "until", where),
*_cite(ln, where),
+ tuple(a.strip().upper() for a in aliases if a.strip()),
+ alias_item,
+ alias_p,
)
)
synonyms = tuple(str(s).strip().lower() for s in raw.get("synonyms") or ())
diff --git a/src/pfs/curated.yaml b/src/pfs/curated.yaml
index 03a90fe..b33728e 100644
--- a/src/pfs/curated.yaml
+++ b/src/pfs/curated.yaml
@@ -31,4 +31,4 @@ HOME-VISITS:
- {kind: add-on, target: G2211, since: 2026, cite: {item_key: NHRGIHGD, p_id: 1469}}
- {kind: modifier, target: B1, since: 2027, cite: {item_key: JCWARTJD, p_id: 1168}}
- {kind: modifier, target: B2, since: 2027, cite: {item_key: JCWARTJD, p_id: 1205}}
- - {kind: prolonged-add-on, target: G0318, since: 2023, cite: {item_key: PETM225D, p_id: 1694}}
+ - {kind: prolonged-add-on, target: G0318, since: 2023, aliases: [GXXX3], alias_cite: {item_key: MZ24MX5S, p_id: 880}, cite: {item_key: PETM225D, p_id: 1694}}
diff --git a/src/pfs/stance.py b/src/pfs/stance.py
index 8d717ec..7e8b714 100644
--- a/src/pfs/stance.py
+++ b/src/pfs/stance.py
@@ -14,6 +14,7 @@ It takes only our own ``StanceRow``s: CMS's summaries of comments
from __future__ import annotations
+import html
import json
import re
from collections import defaultdict
@@ -28,10 +29,14 @@ from pfs.codetables import (
read_events,
read_rounds,
)
+from pfs.descriptors import rule_year_of
+from pfs.families import code_pattern, codes_in
+from pfs.lineage import _expand_ranges
+from pfs.rounds import rule_items
from pfs.runlog import RunLog, input_hash, run_units
TASK = "stance"
-STANCE_PROMPT_VERSION = "stance-v1"
+STANCE_PROMPT_VERSION = "stance-v2"
STANCES = ("support", "modify", "oppose", "unclear")
_EVIDENCE_WORDS = 6
_FIELD_CHARS = 200
@@ -65,8 +70,101 @@ def _num(x: float | None) -> str:
return f"{x:.2f}"
-def docket_events(con: Any, family: Any, rule_year: int) -> list[EventChoice]:
- """The family events a docket's rule touched, always ending in ``other``."""
+def _compact(codes: Sequence[str]) -> str:
+ """ "99324-99328, 99334-99337, 99339, 99340 and 99343": runs of three or
+ more consecutive numeric codes become a range."""
+ out: list[str] = []
+ cs = sorted(set(codes))
+ i = 0
+ while i < len(cs):
+ j = i
+ while (
+ j + 1 < len(cs)
+ and cs[j].isdigit()
+ and cs[j + 1].isdigit()
+ and int(cs[j + 1]) == int(cs[j]) + 1
+ ):
+ j += 1
+ if j - i >= 2:
+ out.append(f"{cs[i]}\u2013{cs[j]}")
+ else:
+ out += cs[i : j + 1]
+ i = j + 1
+ return out[0] if len(out) == 1 else ", ".join(out[:-1]) + " and " + out[-1]
+
+
+def _paragraph(store: Any, item_key: str, p_id: int) -> str:
+ row = (
+ store._con()
+ .execute( # noqa: SLF001
+ "SELECT text FROM fr_anchors WHERE item_key = ? AND p_id = ?",
+ (item_key, p_id),
+ )
+ .fetchone()
+ )
+ return (row[0] or "") if row else ""
+
+
+def _descriptor(para: str, alias: str) -> str:
+ """The descriptor a proposed rule prints for a placeholder code."""
+ m = re.search(
+ rf"\b{re.escape(alias)}\s*\(\s*(.+?)\)\s*\.?\s*\(Do not", para, re.S
+ ) or re.search(rf"\b{re.escape(alias)}\s*\(\s*(.{{1,450}})", para, re.S)
+ if m is None:
+ return ""
+ d = " ".join(m.group(1).split())[:450]
+ return d + ")" * max(0, d.count("(") - d.count(")"))
+
+
+def _link_text(link: Any, rule_year: int, store: Any) -> str:
+ alias = link.aliases[0] if link.aliases else ""
+ desc = ""
+ if alias and store is not None and link.alias_item_key:
+ desc = _descriptor(
+ _paragraph(store, link.alias_item_key, link.alias_p_id), alias
+ )
+ head = f"Linked code {link.target} ({link.kind}), new in the CY{rule_year} rule"
+ if alias:
+ head += f"; the proposed rule called it {alias}"
+ return f"{head}: {desc}" if desc else head
+
+
+def _telehealth_text(family: Any, rule_year: int, store: Any) -> str:
+ """The proposed rule's own last telehealth sentence of the first paragraph
+ that names a family code (or a link's placeholder) together with
+ "telehealth". Empty when there is none."""
+ names = [*family.codes, *(a for ln in family.links for a in ln.aliases)]
+ pattern = code_pattern(names)
+ items = [
+ it
+ for it in rule_items(store, [])
+ if rule_year_of(it["title"], it["date_published"]) == rule_year
+ ]
+ items = [it for it in items if it["kind"] == "proposed"] or items
+ con = store._con() # noqa: SLF001
+ for it in items:
+ for p_id, text in con.execute(
+ "SELECT p_id, text FROM fr_anchors WHERE item_key = ? "
+ "AND text LIKE '%elehealth%' ORDER BY p_id",
+ (it["item_key"],),
+ ).fetchall():
+ if codes_in(_expand_ranges(text or ""), pattern):
+ sentences = [
+ x
+ for x in re.split(r"(?<=[.!?])\s+(?=[A-Z])", " ".join(text.split()))
+ if "elehealth" in x
+ ]
+ return sentences[-1][:500]
+ return ""
+
+
+def docket_events(
+ con: Any, family: Any, rule_year: int, store: Any = None
+) -> list[EventChoice]:
+ """The family events a docket's rule touched, each with a deterministic
+ text of what CMS proposed (R61), always ending in ``other``. *store*
+ (bib) supplies the FR paragraphs behind the telehealth and link events;
+ without it only stored lineage rows are used."""
key = family.key
def ev(slug: str, text: str) -> EventChoice:
@@ -86,40 +184,66 @@ def docket_events(con: Any, family: Any, rule_year: int) -> list[EventChoice]:
None,
)
bits = []
- if ruc is not None:
+ if ruc is not None and ruc == prop:
+ bits.append(f"RUC recommended and CMS proposed {_num(ruc)}")
+ elif ruc is not None:
bits.append(f"RUC recommended {_num(ruc)}")
- if prop is not None:
+ if prop is not None and ruc != prop:
bits.append(f"CMS proposed {_num(prop)}")
- parts.append(f"{code} ({', '.join(bits)})" if bits else code)
- out.append(ev("valuation", "Work RVU valuation: " + "; ".join(parts)))
+ parts.append(f"{code} ({'; '.join(bits)})" if bits else code)
+ out.append(
+ ev(
+ "valuation",
+ f"Work RVU valuation in the CY{rule_year} rule: " + ", ".join(parts),
+ )
+ )
events = [
e for c in family.codes for e in read_events(con, c) if e.year == rule_year
]
- coding = sorted({(e.code, e.kind) for e in events if e.kind in _CODING_KINDS})
- if coding:
- out.append(
- ev(
- "coding",
- "Coding changes (CPT codes created, deleted, revised or acted on by "
- "the CPT Editorial Panel): "
- + "; ".join(f"{c} {k.replace('_', ' ')}" for c, k in coding),
+ deleted = sorted(m.code for m in family.members if m.until == rule_year - 1)
+ panel = sorted({(e.code, e.kind) for e in events if e.kind in _CODING_KINDS})
+ if deleted or panel:
+ bits = []
+ if deleted:
+ kept = sorted(
+ m.code
+ for m in family.members
+ if m.role == "base" and (m.since or 0) <= rule_year
)
- )
- tele = sorted({e.code for e in events if e.kind == "telehealth_list"})
- if tele:
- out.append(
- ev("telehealth", "Medicare telehealth services list: " + ", ".join(tele))
- )
- for link in family.links:
- if link.since == rule_year:
- out.append(
- ev(
- link.target,
- f"Linked code {link.target} ({link.kind}) introduced with "
- f"this family in CY{rule_year}",
+ bits.append(
+ f"CPT deleted {_compact(deleted)}; those services are reported with "
+ f"{_compact(kept)} from CY{rule_year}"
+ )
+ if panel:
+ kinds = sorted({k for _, k in panel})
+ bits.append(
+ "CPT Editorial Panel actions on family codes that year: "
+ + ", ".join(
+ f"{sum(1 for _, k2 in panel if k2 == k)} {k.replace('_', ' ')}"
+ for k in kinds
)
)
- out.append(ev("other", "Any other aspect of the family or its payment"))
+ out.append(ev("coding", "Coding changes: " + ". ".join(bits)))
+ tele = sorted({e.code for e in events if e.kind == "telehealth_list"})
+ para = _telehealth_text(family, rule_year, store) if store is not None else ""
+ if tele or para:
+ text = "Medicare telehealth services list"
+ if para:
+ text += f", from the proposed rule: {para}"
+ if tele:
+ text += f" (lineage: telehealth list change for {', '.join(tele)})"
+ out.append(ev("telehealth", text))
+ for link in family.links:
+ if link.since == rule_year:
+ out.append(ev(link.target, _link_text(link, rule_year, store)))
+ out.append(
+ ev(
+ "other",
+ "Any other aspect of this family's own codes "
+ f"({_compact([m.code for m in family.members if m.until is None])}), "
+ "not covered above; positions on other codes or services are not given",
+ )
+ )
return out
@@ -151,28 +275,124 @@ def stance_schema(events: Sequence[EventChoice]) -> dict:
def _system(events: Sequence[EventChoice]) -> str:
listed = "\n".join(f"- {e.event_key}: {e.text}" for e in events)
return (
- "You read one public comment on a Medicare payment rule and say what the "
- "commenter's position is on each of these events, only for events the "
- "comment actually addresses:\n"
+ "You read one public comment on a Medicare payment rule. Below are the "
+ "events of one code family that the proposed rule touched, each with what "
+ "CMS proposed:\n"
f"{listed}\n"
- "For each position give event (one of the keys above), stance (support, "
- "modify when it asks for a change, oppose, or unclear), reason (one short "
- "sentence) and evidence (an exact quote of at least six words copied from "
- "the comment, or an empty string). Give no position for an event the "
- "comment does not address."
+ "Give the commenter's position on an event only when the comment itself "
+ "takes one on that event, even in a single sentence (the commenter may use "
+ "a placeholder code or a short name for it); never guess or infer a "
+ "position. stance: support = the "
+ "commenter endorses what CMS proposed for that event as written (we "
+ "support, we agree, we appreciate, finalize as proposed); oppose = the "
+ "commenter disagrees with it (does not support, urges CMS not to finalize, "
+ "says it fails its goal); modify = the commenter asks for any change, "
+ "addition, delay, retention or clarification that goes beyond or differs "
+ "from what CMS proposed, even while welcoming the idea (we recommend, we "
+ "urge CMS to add, reconsider, clarify); unclear = the comment addresses the "
+ "event but the stance cannot be determined. If the comment does not address "
+ "an event, give no position at all for it; most comments address only one or "
+ "two events. Give at most one position per event, and never use the same "
+ "quote for two events. A comment about other codes or other specialties' services "
+ "(audiology, therapy, surgery) addresses none of these events. reason: one "
+ "short sentence. evidence: an exact quote of at least six words copied from "
+ "the comment that shows the position."
)
+def clean_text(text: str) -> str:
+ """The comment as plain text: HTML entities decoded, ``
`` to a
+ newline (regulations.gov stores inline comments with both)."""
+ return re.sub(r"", "\n", html.unescape(text or ""))
+
+
def _words(s: str) -> list[str]:
return re.findall(r"[a-z0-9]+", (s or "").lower())
-def _quoted(evidence: str, text_words: str) -> bool:
- ws = _words(evidence)
- return any(
- f" {' '.join(ws[i : i + _EVIDENCE_WORDS])} " in text_words
- for i in range(len(ws) - _EVIDENCE_WORDS + 1)
- )
+def evidence_span(quote: str, text: str) -> str:
+ """The longest verbatim run of the comment's own words that *quote*
+ matches (as the origin dossier's longest span), at least six words and
+ at most ``_FIELD_CHARS`` characters; empty when there is none."""
+ toks = text.split()
+ flat: list[str] = []
+ owner: list[int] = []
+ for i, t in enumerate(toks):
+ for w in _words(t):
+ flat.append(w)
+ owner.append(i)
+ q = _words(quote)
+ best, end = 0, 0
+ prev = [0] * (len(q) + 1)
+ for i, w in enumerate(flat):
+ cur = [0] * (len(q) + 1)
+ for j, qw in enumerate(q):
+ if w == qw:
+ cur[j + 1] = prev[j] + 1
+ if cur[j + 1] > best:
+ best, end = cur[j + 1], i
+ prev = cur
+ if best < _EVIDENCE_WORDS:
+ return ""
+ span = " ".join(toks[owner[end - best + 1] : owner[end] + 1])
+ return span if len(span) <= _FIELD_CHARS else span[:_FIELD_CHARS].rsplit(" ", 1)[0]
+
+
+def raw_positions(
+ reply: dict | None, events: Sequence[EventChoice]
+) -> list[dict] | None:
+ """The reply's valid positions, all of them: known event, known stance.
+ This is what the ledger keeps (R61 f)."""
+ if reply is None:
+ return None
+ known = {e.event_key for e in events}
+ out = []
+ for p in reply.get("positions") or []:
+ if not isinstance(p, dict) or p.get("event") not in known:
+ continue
+ if p.get("stance") not in STANCES:
+ continue
+ out.append(
+ {
+ "event": p["event"],
+ "stance": p["stance"],
+ "reason": str(p.get("reason") or "")[:_FIELD_CHARS],
+ "evidence": str(p.get("evidence") or "")[:_FIELD_CHARS],
+ }
+ )
+ return out
+
+
+def position_rows(
+ raw: Sequence[dict],
+ text: str,
+ *,
+ family: str,
+ item_key: str,
+ docket: str,
+ extractor: str,
+) -> list[StanceRow]:
+ """The kept positions of a raw reply: evidence becomes the longest
+ verbatim span of *text* (>= 6 words) or the position is dropped, then
+ the first position per event is kept, and one span backs one position
+ only (a quote copied onto several events is kept for the first)."""
+ rows: dict[str, StanceRow] = {}
+ used: set[str] = set()
+ for p in raw:
+ span = evidence_span(p["evidence"], text)
+ if span and p["event"] not in rows and span not in used:
+ used.add(span)
+ rows[p["event"]] = StanceRow(
+ family,
+ item_key,
+ docket,
+ p["event"],
+ p["stance"],
+ p["reason"],
+ span,
+ extractor,
+ )
+ return list(rows.values())
def classify_comment(
@@ -185,87 +405,63 @@ def classify_comment(
docket: str,
) -> list[StanceRow] | None:
"""Positions of one comment, or ``None`` when the model gave no usable
- reply. A position on an unknown event or with an unknown stance is
- dropped; ``evidence`` is kept only if >= 6 consecutive words of it occur
- in *text*."""
- reply = ask(_system(events), text, stance_schema(events))
- if reply is None:
+ reply."""
+ text = clean_text(text)
+ raw = raw_positions(ask(_system(events), text, stance_schema(events)), events)
+ if raw is None:
return None
- known = {e.event_key for e in events}
- hay = f" {' '.join(_words(text))} "
extractor = f"model:{getattr(ask, 'model', 'unknown')}@{STANCE_PROMPT_VERSION}"
- rows = []
- for p in reply.get("positions") or []:
- if not isinstance(p, dict) or p.get("event") not in known:
- continue
- if p.get("stance") not in STANCES:
- continue
- evidence = str(p.get("evidence") or "")
- rows.append(
- StanceRow(
- family,
- item_key,
- docket,
- p["event"],
- p["stance"],
- str(p.get("reason") or "")[:_FIELD_CHARS],
- evidence[:_FIELD_CHARS] if _quoted(evidence, hay) else "",
- extractor,
- )
- )
- return rows
+ return position_rows(
+ raw, text, family=family, item_key=item_key, docket=docket, extractor=extractor
+ )
# ── run ──────────────────────────────────────────────────────────────
-def _as_dict(r: StanceRow) -> dict:
- return {
- "event": r.event_key,
- "stance": r.stance,
- "reason": r.reason,
- "evidence": r.evidence,
- }
-
-
def stances_for(
commenters: Sequence[CommenterRow],
texts: Mapping[str, str],
events: Sequence[EventChoice],
- ask: Callable,
+ ask: Callable | None,
log: RunLog,
*,
family: str,
docket: str,
+ model: str | None = None,
limit: int = 0,
retry_failed: bool = False,
progress: Callable[[str, str], None] | None = None,
) -> tuple[list[StanceRow], dict[str, int]]:
- """Classify each commenter's comment through the resumable ledger.
- Counts are the run counts plus ``pending`` (no ledger answer yet) and
- ``no_text`` (a commenter row whose comment text is not found)."""
- model = getattr(ask, "model", "unknown")
+ """Classify each commenter's comment through the resumable ledger, which
+ keeps the validated raw reply; the evidence and one-per-event filters run
+ here on replay. With *ask* ``None`` (``--no-llm``) only the ledger of
+ *model* is read. Counts are the run counts plus ``pending`` (no ledger
+ answer yet) and ``no_text`` (a commenter whose comment text is not found)."""
+ model = model or getattr(ask, "model", "unknown")
extractor = f"model:{model}@{STANCE_PROMPT_VERSION}"
- ev_part = [f"{e.event_key}|{e.text}" for e in events]
+ prompt = _system(events) # events and rules both key the ledger
keys = [c.item_key for c in commenters if c.item_key in texts]
- hashes = {k: input_hash(texts[k], *ev_part) for k in keys}
+ clean = {k: clean_text(texts[k]) for k in keys}
+ hashes = {k: input_hash(clean[k], prompt) for k in keys}
+ counts = {"done": 0, "skipped": 0, "failed": 0, "failed_pending": 0}
+ if ask is not None:
- def unit(k: str) -> dict | None:
- rows = classify_comment(
- texts[k], events, ask, family=family, item_key=k, docket=docket
+ def unit(k: str) -> dict | None:
+ raw = raw_positions(
+ ask(_system(events), clean[k], stance_schema(events)), events
+ )
+ return None if raw is None else {"positions": raw}
+
+ counts = run_units(
+ [(k, hashes[k], lambda k=k: unit(k)) for k in keys],
+ log,
+ model=model,
+ version=STANCE_PROMPT_VERSION,
+ limit=limit,
+ retry_failed=retry_failed,
+ progress=progress,
)
- return None if rows is None else {"positions": [_as_dict(r) for r in rows]}
-
- counts = run_units(
- [(k, hashes[k], lambda k=k: unit(k)) for k in keys],
- log,
- model=model,
- version=STANCE_PROMPT_VERSION,
- limit=limit,
- retry_failed=retry_failed,
- progress=progress,
- )
- known = {e.event_key for e in events}
rows: list[StanceRow] = []
pending = 0
for k in keys:
@@ -273,47 +469,50 @@ def stances_for(
if entry is None or entry.output is None:
pending += 1
continue
- rows += [
- StanceRow(
- family,
- k,
- docket,
- p["event"],
- p["stance"],
- p["reason"],
- p["evidence"],
- extractor,
- )
- for p in entry.output["positions"]
- if p["event"] in known
- ]
+ rows += position_rows(
+ entry.output["positions"],
+ clean[k],
+ family=family,
+ item_key=k,
+ docket=docket,
+ extractor=extractor,
+ )
return rows, {**counts, "pending": pending, "no_text": len(commenters) - len(keys)}
# ── roll-up ──────────────────────────────────────────────────────────
-def _rank(event_key: str) -> tuple[int, str]:
- slug = event_key.rsplit(":", 1)[-1]
- order = {"valuation": 0, "coding": 1, "telehealth": 2, "other": 99}
- return order.get(slug, 3), event_key
-
-
def _first_firm(stances: Sequence[str]) -> str:
return next((s for s in stances if s != "unclear"), "unclear")
-def rollup(stances: Sequence[StanceRow], *, base: ReactionRow) -> ReactionRow:
+def rollup(
+ stances: Sequence[StanceRow],
+ *,
+ base: ReactionRow,
+ order: Sequence[str] = (),
+ n_commenters: int | None = None,
+) -> ReactionRow:
"""Distinct commenters by their first non-``unclear`` position in event
order (all ``unclear`` -> ``unclear``); ``events_json`` counts distinct
- commenters per event the same way. Only our own stance rows go in."""
+ commenters per event the same way. *order* is the docket's event keys in
+ ``docket_events`` order (an event not in it sorts last, by key).
+ ``n_classified`` is the commenters with a firm position and
+ ``n_commenters`` the commenter rows of the docket (R62). Only our own
+ stance rows go in; CMS's summaries never do."""
+ rank = {k: i for i, k in enumerate(order)}
+
+ def key(e: str) -> tuple[int, str]:
+ return rank.get(e, len(rank)), e
+
by_item: dict[str, dict[str, list[str]]] = defaultdict(lambda: defaultdict(list))
for s in stances:
by_item[s.item_key][s.event_key].append(s.stance)
totals = dict.fromkeys(STANCES, 0)
per_event: dict[str, dict[str, int]] = {}
for events in by_item.values():
- ordered = sorted(events, key=_rank)
+ ordered = sorted(events, key=key)
totals[_first_firm([_first_firm(events[e]) for e in ordered])] += 1
for e in ordered:
per_event.setdefault(e, dict.fromkeys(STANCES, 0))[
@@ -326,6 +525,8 @@ def rollup(stances: Sequence[StanceRow], *, base: ReactionRow) -> ReactionRow:
stance_oppose=totals["oppose"],
stance_unclear=totals["unclear"],
events_json=json.dumps(
- dict(sorted(per_event.items(), key=lambda kv: _rank(kv[0])))
+ dict(sorted(per_event.items(), key=lambda kv: key(kv[0])))
),
+ n_classified=len(by_item) - totals["unclear"],
+ n_commenters=n_commenters,
)
diff --git a/tests/cli/test_pfs_cli.py b/tests/cli/test_pfs_cli.py
index 1ec693e..7849c2b 100644
--- a/tests/cli/test_pfs_cli.py
+++ b/tests/cli/test_pfs_cli.py
@@ -2,6 +2,7 @@
from __future__ import annotations
+import json
from pathlib import Path
from unittest.mock import MagicMock
@@ -1969,6 +1970,17 @@ class TestStance:
monkeypatch.setattr(pfs_cli, "_store", lambda: store)
monkeypatch.setattr(pfs_cli, "_engine", lambda: "ENGINE")
monkeypatch.setattr(pfs_cli, "_asker", lambda: "ASK")
+ monkeypatch.setattr(pfs_cli, "_model_name", lambda: "qwen-x")
+ from pfs.stance import EventChoice
+
+ monkeypatch.setattr(
+ pfs_cli,
+ "docket_events",
+ lambda con_, fam, year, store_: [
+ EventChoice("HOME-VISITS:2023:G0318", "g", year),
+ EventChoice("HOME-VISITS:2023:other", "o", year),
+ ],
+ )
monkeypatch.setattr(
pfs_cli, "ledger_path", lambda t, f: Path("/nonexistent") / f"{t}-{f}"
)
@@ -2040,11 +2052,21 @@ class TestStance:
rows = {r.period: r for r in read_reaction(con._real, "HOME-VISITS")}
assert set(rows) == {"D1", "D0", "R1"}
assert (rows["D1"].stance_support, rows["D1"].stance_oppose) == (1, 1)
+ assert (rows["D1"].n_classified, rows["D1"].n_commenters) == (2, 2)
+ assert (rows["D0"].n_classified, rows["D0"].n_commenters) == (None, None)
+ assert list(json.loads(rows["D1"].events_json)) == ["HOME-VISITS:2023:other"]
assert rows["D1"].n_items == 2 and rows["D1"].n_total == 9
assert (rows["D0"].stance_support, rows["R1"].stance_oppose) == (0, 0)
assert rows["D0"].events_json == "{}"
assert con.published == [True]
+ def test_no_llm_replays_the_ledger(self, con, monkeypatch):
+ seen = self._patch(con, monkeypatch)
+ monkeypatch.setattr(pfs_cli, "_asker", lambda: 1 / 0)
+ res = runner.invoke(app, self._args("--no-llm"))
+ assert res.exit_code == 0, res.output
+ assert seen["ask"] is None and seen["model"] == "qwen-x"
+
def test_write_with_no_stored_reaction_rows_uses_series(self, con, monkeypatch):
self._patch(con, monkeypatch)
res = runner.invoke(app, self._args("--write"))
@@ -2145,6 +2167,22 @@ class TestStance:
from pfs.codetables import StanceRow, write_stances
self._reaction_patch(monkeypatch)
+ from pfs.codetables import CommenterRow, write_commenters, write_reaction
+
+ write_commenters(
+ con._real,
+ [
+ CommenterRow(k, "D1", k, "", "individual", "", "", 0, "regs_gov_fields", "", "regs", "CCM")
+ for k in ("K1", "K2", "K3")
+ ],
+ ) # fmt: skip
+ write_reaction(
+ con._real,
+ "CCM",
+ [
+ ReactionRow("CCM", "D1", "docket", 2023, 5, 20, 0, 0, 0, 0, "[]", '{"CCM:2023:other": {}}')
+ ],
+ ) # fmt: skip
write_stances(
con._real,
"CCM",
@@ -2158,6 +2196,9 @@ class TestStance:
assert res.exit_code == 0, res.output
rows = {r.period: r for r in read_reaction(con._real, "CCM")}
assert rows["D1"].stance_modify == 2 and rows["D1"].n_items == 5
+ assert (rows["D1"].n_classified, rows["D1"].n_commenters) == (2, 3)
+ assert (rows["D2"].n_classified, rows["D2"].n_commenters) == (None, None)
+ assert list(json.loads(rows["D1"].events_json)) == ["CCM:2023:other"]
assert rows["D2"].stance_modify == 0 and rows["D2"].events_json == "{}"
def test_stance_sample_refused_when_stances_stored(self, con, monkeypatch):
diff --git a/tests/pfs/test_curated.py b/tests/pfs/test_curated.py
index cf83f65..b66e16e 100644
--- a/tests/pfs/test_curated.py
+++ b/tests/pfs/test_curated.py
@@ -194,3 +194,39 @@ def test_home_visits_pair_phrases_are_visit_naming():
"home visits" not in fam.pair_phrases and "domiciliary" not in fam.pair_phrases
)
assert "house call" in fam.pair_phrases
+
+
+def test_link_aliases_and_their_cite():
+ fam = load_curated()["HOME-VISITS"]
+ g = next(link for link in fam.links if link.target == "G0318")
+ assert (g.aliases, g.alias_item_key, g.alias_p_id) == (("GXXX3",), "MZ24MX5S", 880)
+ b = next(link for link in fam.links if link.target == "B1")
+ assert (b.aliases, b.alias_item_key, b.alias_p_id) == ((), "", 0)
+
+
+def test_alias_parsing_and_errors(tmp_path):
+ head = "F:\n name: F\n links:\n - {kind: add-on, target: G1, since: 2023, "
+ ok = (
+ head
+ + "aliases: [' gxxx3 ', ''], alias_cite: {item_key: K, p_id: 5}, cite: {item_key: K, p_id: 1}}\n"
+ )
+ link = load_curated(_write(tmp_path, ok))["F"].links[0]
+ assert link.aliases == ("GXXX3",) and link.alias_p_id == 5
+ for bad, needle in [
+ (
+ head + "aliases: GXXX3, cite: {item_key: K, p_id: 1}}\n",
+ "aliases must be a list",
+ ),
+ (
+ head + "aliases: [1], cite: {item_key: K, p_id: 1}}\n",
+ "aliases must be a list",
+ ),
+ (head + "aliases: [X], cite: {item_key: K, p_id: 1}}\n", "alias_cite"),
+ (
+ head
+ + "aliases: [X], alias_cite: {item_key: K, page: 1}, cite: {item_key: K, p_id: 1}}\n",
+ "alias_cite",
+ ),
+ ]:
+ with pytest.raises(ValueError, match=needle):
+ load_curated(_write(tmp_path, bad))
diff --git a/tests/pfs/test_reaction.py b/tests/pfs/test_reaction.py
index db7ef21..412c8e3 100644
--- a/tests/pfs/test_reaction.py
+++ b/tests/pfs/test_reaction.py
@@ -394,9 +394,15 @@ class TestEventsJson:
ev = json.dumps(
{"F:2023:other": {"support": 1, "modify": 0, "oppose": 0, "unclear": 0}}
)
- row = ReactionRow("CCM", "D1", "docket", 2023, 5, 20, 1, 0, 0, 0, "[]", ev)
+ row = ReactionRow(
+ "CCM", "D1", "docket", 2023, 5, 20, 1, 0, 0, 0, "[]", ev, 4, 9
+ )
write_reaction(con, "CCM", [row])
assert read_reaction(con, "CCM") == [row]
+ never = ReactionRow("CCM", "D2", "docket", 2024, 1, 2, 0, 0, 0, 0, "[]")
+ assert (never.n_classified, never.n_commenters) == (None, None)
+ write_reaction(con, "CCM", [row, never])
+ assert read_reaction(con, "CCM")[1].n_classified is None
assert (
ReactionRow("CCM", "D", "docket", 1, 0, 0, 0, 0, 0, 0, "[]").events_json
== "{}"
@@ -408,6 +414,7 @@ class TestEventsJson:
"sample_json) VALUES ('OLD', 'D0', 'docket', 2020, 1, 2, 0, 0, 0, 0, '[]')"
)
assert read_reaction(con, "OLD")[0].events_json == "{}"
+ assert read_reaction(con, "OLD")[0].n_commenters is None
def test_ensure_tables_adds_the_column_to_an_old_table(self):
c = duckdb.connect(":memory:")
@@ -421,8 +428,15 @@ class TestEventsJson:
c.execute(
"INSERT INTO pfs.code_reaction VALUES ('A','D','docket',1,1,1,0,0,0,0,'[]')"
)
+ short = read_reaction(c, "A")[0] # an old replica: rows without the columns
+ assert (short.events_json, short.n_classified) == ("{}", None)
ensure_tables(c)
- assert read_reaction(c, "A")[0].events_json == "{}"
+ got = read_reaction(c, "A")[0]
+ assert (got.events_json, got.n_classified, got.n_commenters) == (
+ "{}",
+ None,
+ None,
+ )
class TestFrPairSpansRefactor:
diff --git a/tests/pfs/test_stance.py b/tests/pfs/test_stance.py
index 938d113..87da1be 100644
--- a/tests/pfs/test_stance.py
+++ b/tests/pfs/test_stance.py
@@ -4,11 +4,12 @@ from __future__ import annotations
import dataclasses
import json
-from types import SimpleNamespace
import duckdb
import pytest
+from bib.item import Rule
+from bib.store import Store
from pfs.codetables import (
CmsSummaryRow,
CommenterRow,
@@ -23,14 +24,18 @@ from pfs.codetables import (
write_rounds,
write_stances,
)
-from pfs.curated import CuratedLink
+from pfs.curated import load_curated
from pfs.runlog import RunLog
from pfs.stance import (
STANCE_PROMPT_VERSION,
EventChoice,
classify_comment,
+ clean_text,
docket_events,
docket_rule_year,
+ evidence_span,
+ position_rows,
+ raw_positions,
rollup,
stance_schema,
stances_for,
@@ -44,14 +49,7 @@ TEXT = (
)
EV_VAL = "we support the proposed increase in work RVUs for CPT code 99342"
EV_G = "we urge CMS to reconsider the new G0318 prolonged services add-on"
-FAM = SimpleNamespace(
- key="HOME-VISITS",
- codes=("99341", "99342"),
- links=(
- CuratedLink("prolonged-add-on", "G0318", 2023, None, "PETM225D", 1694, 0, 0),
- CuratedLink("add-on", "G2211", 2026, None, "NHRGIHGD", 1469, 0, 0),
- ),
-)
+FAM = load_curated()["HOME-VISITS"]
def _round(code="99342", year=2023):
@@ -120,25 +118,90 @@ def test_docket_rule_year_matches_rag():
# ── events ───────────────────────────────────────────────────────────
+TELE_PARA = (
+ "For home or residence visits CPT codes 99345 and 99350 apply. "
+ "These services are similar to services currently on the Medicare Telehealth "
+ "Services List, such as CPT codes 99356 and 99357. Similarly, we believe that "
+ "these proposed HCPCS G codes would qualify. Therefore, we are proposing to add "
+ "proposed HCPCS codes GXXX1, GXXX2, and GXXX3 to the Medicare Telehealth "
+ "Services List on a Category 1 basis."
+)
+ALIAS_PARA = (
+ "We are proposing that prolonged home or residence services would be reportable "
+ "under GXXX3 ( Prolonged home or residence evaluation and management service(s) "
+ "beyond the total time for the primary service; each additional 15 minutes "
+ "(list separately in addition to CPT codes 99345, 99350 for home or residence "
+ "evaluation and management services). (Do not report GXXX3 on the same date "
+ "as 99358))."
+)
-def test_docket_events_cy2023_home_visits(con):
+
+def _store(title, paras):
+ s = Store(":memory:")
+ key = s.create(Rule(title=title))
+ c = s._con() # noqa: SLF001
+ c.executemany(
+ "INSERT INTO fr_anchors (item_key, p_id, page, ordinal, text) VALUES (?,?,?,?,?)",
+ [(key, p, 1, p, t) for p, t in paras.items()],
+ )
+ c.commit()
+ return s, key
+
+
+@pytest.fixture
+def home_store():
+ s, key = _store(
+ "CY 2023 Payment Policies Under the Physician Fee Schedule; Proposed Rule",
+ {5: TELE_PARA, 6: "An unrelated paragraph about 99342.", 880: ALIAS_PARA},
+ )
+ yield s, key
+ s.close()
+
+
+def _fam(key):
+ links = tuple(
+ dataclasses.replace(ln, alias_item_key=key) if ln.aliases else ln
+ for ln in FAM.links
+ )
+ return dataclasses.replace(FAM, links=links)
+
+
+def test_docket_events_cy2023_home_visits(con, home_store):
+ store, key = home_store
write_rounds(con, "HOME-VISITS", [_round(), _round(year=2008)])
write_events(con, "99342", [_event("99342", 2023, "cpt_panel")])
- got = docket_events(con, FAM, 2023)
+ fam = _fam(key)
+ got = docket_events(con, fam, 2023, store)
assert [e.event_key for e in got] == [
"HOME-VISITS:2023:valuation",
"HOME-VISITS:2023:coding",
+ "HOME-VISITS:2023:telehealth",
"HOME-VISITS:2023:G0318",
"HOME-VISITS:2023:other",
]
assert all(e.rule_year == 2023 for e in got)
- val = got[0].text
- assert "99342" in val and "RUC" in val and "2.00" in val and "1.90" in val
- assert "2008" not in val
- assert docket_events(con, FAM, 2023) == got # deterministic
+ val, coding, tele, g, other = (e.text for e in got)
+ assert (
+ "99342 (RUC recommended 2.00; CMS proposed 1.90)" in val and "2008" not in val
+ )
+ assert (
+ "CPT deleted 99324–99328, 99334–99337, 99339, 99340 and 99343; those "
+ "services are reported with 99341, 99342, 99344, 99345 and 99347–99350 "
+ "from CY2023" in coding
+ )
+ assert "1 cpt panel" in coding
+ assert tele.endswith(
+ "GXXX3 to the Medicare Telehealth Services List on a Category 1 basis."
+ )
+ assert "99356" not in tele and "unrelated" not in tele
+ assert g.startswith("Linked code G0318 (prolonged-add-on), new in the CY2023 rule")
+ assert "called it GXXX3: Prolonged home or residence evaluation" in g
+ assert g.endswith("home or residence evaluation and management services)")
+ assert "99341" in other and "other codes or services are not given" in other
+ assert docket_events(con, fam, 2023, store) == got # deterministic
-def test_docket_events_telehealth_and_bare_year(con):
+def test_docket_events_without_store_uses_lineage_only(con):
write_events(
con,
"99341",
@@ -150,19 +213,75 @@ def test_docket_events_telehealth_and_bare_year(con):
"telehealth",
"other",
]
+ assert "1 created" in got[0].text and "deleted" not in got[0].text
+ assert got[1].text.endswith("(lineage: telehealth list change for 99341)")
assert [e.event_key for e in docket_events(con, FAM, 1999)] == [
"HOME-VISITS:1999:other"
]
+def test_docket_events_link_without_alias_or_paragraph(con):
+ got = docket_events(con, FAM, 2026)
+ assert got[0].event_key == "HOME-VISITS:2026:G2211"
+ assert got[0].text == "Linked code G2211 (add-on), new in the CY2026 rule"
+ store = Store(":memory:") # alias paragraph absent
+ g = docket_events(con, FAM, 2023, store)[-2]
+ assert g.text.endswith("the proposed rule called it GXXX3")
+
+
+def test_telehealth_paragraph_with_a_range_and_item_fallback(con):
+ s, _ = _store(
+ "CY 2024 Payment Policies Under the Physician Fee Schedule; Final Rule",
+ {
+ 1: "Telehealth is covered for CPT codes 99324-99328 as proposed.",
+ 2: "Telehealth for something else, 12345.",
+ },
+ )
+ got = docket_events(con, FAM, 2024, s)
+ assert [e.event_key.split(":")[-1] for e in got] == ["telehealth", "other"]
+ assert "99324-99328" in got[0].text
+ s.close()
+ s2, _ = _store("CY 2024 Payment Policies; PFS", {1: "Telehealth for 12345."})
+ assert len(docket_events(con, FAM, 2024, s2)) == 1
+ assert len(docket_events(con, FAM, 2030, s2)) == 1
+ s2.close()
+
+
+def test_descriptor_without_do_not_clause(con):
+ s, key = _store(
+ "CY 2023 Payment Policies Under the Physician Fee Schedule; Proposed Rule",
+ {880: "We propose GXXX3 ( Prolonged home visit (each 15 minutes) of services."},
+ )
+ g = docket_events(con, _fam(key), 2023, s)[-2]
+ assert (
+ "called it GXXX3: Prolonged home visit (each 15 minutes) of services." in g.text
+ )
+ s.close()
+ s3, key3 = _store(
+ "CY 2023 Payment Policies Under the Physician Fee Schedule; Proposed Rule",
+ {880: "no placeholder here"},
+ )
+ assert docket_events(con, _fam(key3), 2023, s3)[-2].text.endswith("called it GXXX3")
+ s3.close()
+
+
def test_docket_events_round_without_values(con):
- r = _round()
r = dataclasses.replace(
- r, ruc_work_rvu=None, proposed_work_rvu=None, final_work_rvu=None
+ _round(), ruc_work_rvu=None, proposed_work_rvu=None, final_work_rvu=None
)
write_rounds(con, "HOME-VISITS", [r])
val = docket_events(con, FAM, 2023)[0]
- assert "99342" in val.text and "RUC" not in val.text
+ assert val.text.endswith("99342") and "RUC" not in val.text
+ r3 = dataclasses.replace(_round(), ruc_work_rvu=1.9)
+ write_rounds(con, "HOME-VISITS", [r3])
+ assert docket_events(con, FAM, 2023)[0].text.endswith(
+ "99342 (RUC recommended and CMS proposed 1.90)"
+ )
+ r2 = dataclasses.replace(_round(), proposed_work_rvu=None)
+ write_rounds(con, "HOME-VISITS", [r2])
+ assert docket_events(con, FAM, 2023)[0].text.endswith(
+ "99342 (RUC recommended 2.00)"
+ )
def test_stance_schema(events):
@@ -177,7 +296,7 @@ def test_stance_schema(events):
]
assert item["properties"]["reason"]["maxLength"] == 200
assert item["properties"]["evidence"]["maxLength"] == 200
- assert STANCE_PROMPT_VERSION == "stance-v1"
+ assert STANCE_PROMPT_VERSION == "stance-v2"
# ── classify_comment ─────────────────────────────────────────────────
@@ -190,11 +309,20 @@ def _classify(positions, text=TEXT, events=None, ask=None):
)
-def test_prompt_lists_events_and_comment(events):
+def test_prompt_lists_events_and_rules(events):
ask = Ask([])
_classify([], events=events, ask=ask)
system, user, schema = ask.calls[0]
- assert "HOME-VISITS:2023:G0318" in system and "New add-on G0318" in system
+ assert "HOME-VISITS:2023:G0318: New add-on G0318" in system
+ for rule in (
+ "support = the commenter endorses what CMS proposed",
+ "modify = the commenter asks for any change",
+ "oppose = the commenter disagrees",
+ "give no position at all",
+ "at most one position per event",
+ "other codes or other specialties",
+ ):
+ assert rule in system
assert user == TEXT
assert schema == stance_schema(events)
@@ -216,26 +344,29 @@ def test_rows_carry_fields(events):
"HOME-VISITS:2023:valuation",
"support",
"likes increase",
- EV_VAL,
- "model:qwen-test@stance-v1",
+ "We support the proposed increase in work RVUs for CPT code 99342",
+ "model:qwen-test@stance-v2",
)
]
def test_unknown_event_in_reply_dropped(events):
rows = _classify(
- [pos("HOME-VISITS:2023:nope", "support"), pos(events[3].event_key, "oppose")],
+ [
+ pos("HOME-VISITS:2023:nope", "support", "r", EV_VAL),
+ pos(events[3].event_key, "oppose", "r", EV_VAL),
+ ],
events=events,
)
assert [r.event_key for r in rows] == [events[3].event_key]
def test_bad_stance_and_non_dict_dropped(events):
- ask = Ask([pos(events[0].event_key, "maybe"), "junk"])
+ ask = Ask([pos(events[0].event_key, "maybe", "r", EV_VAL), "junk"])
assert _classify(None, events=events, ask=ask) == []
-def test_evidence_must_occur_in_text(events):
+def test_evidence_must_occur_in_text_else_position_dropped(events):
rows = _classify(
[
pos(events[0].event_key, "support", "r", EV_VAL),
@@ -245,19 +376,109 @@ def test_evidence_must_occur_in_text(events):
],
events=events,
) # fmt: skip
- assert [r.evidence for r in rows] == [EV_VAL, "", "", ""]
- assert [r.stance for r in rows] == ["support", "oppose", "modify", "unclear"]
+ assert [(r.event_key, r.stance) for r in rows] == [(events[0].event_key, "support")]
+
+
+def test_evidence_is_the_longest_verbatim_span():
+ text = "Intro words here. We urge CMS to reconsider the new G0318 prolonged services add-on code, which will be hard."
+ q = "CMS to reconsider the new G0318 prolonged services add-on code, and more"
+ # the matched run is cut at the quote's own words, in the comment's own spelling
+ assert (
+ evidence_span(q, text)
+ == "CMS to reconsider the new G0318 prolonged services add-on code,"
+ )
+ assert (
+ evidence_span("nothing in common whatsoever with the text at all", text) == ""
+ )
+ assert evidence_span("", text) == ""
+
+
+def test_evidence_span_capped_at_200_chars():
+ text = " ".join(f"word{i}" for i in range(80))
+ span = evidence_span(text, text)
+ assert 0 < len(span) <= 200 and text.startswith(span) and not span.endswith(" ")
def test_evidence_window_match_is_case_and_punctuation_blind(events):
ev = "WE SUPPORT, the proposed increase: in work RVUs for CPT code 99342 (and more)"
rows = _classify([pos(events[0].event_key, "support", "r", ev)], events=events)
- assert rows[0].evidence == ev
+ assert rows[0].evidence.startswith("We support the proposed increase in work RVUs")
-def test_reason_and_evidence_truncated(events):
- rows = _classify([pos(events[0].event_key, "support", "x" * 500)], events=events)
+def test_clean_text_decodes_html():
+ assert (
+ clean_text("people's "x"
next
end")
+ == 'people\'s "x"\nnext\nend'
+ )
+ assert clean_text(None) == ""
+ rows = _classify(
+ [
+ pos(
+ "HOME-VISITS:2023:other",
+ "support",
+ "r",
+ "this is what the people's own words say here",
+ )
+ ],
+ text="this is what the people's own words say here",
+ events=[EventChoice("HOME-VISITS:2023:other", "o", 2023)],
+ )
+ assert rows[0].evidence == "this is what the people's own words say here"
+
+
+def test_one_position_per_event_first_with_evidence_wins(events):
+ k = events[0].event_key
+ rows = _classify(
+ [
+ pos(k, "oppose", "first, no evidence", "nothing like the text of this comment here"),
+ pos(k, "support", "second", EV_VAL),
+ pos(k, "modify", "third", EV_VAL),
+ ],
+ events=events,
+ ) # fmt: skip
+ assert [(r.stance, r.reason) for r in rows] == [("support", "second")]
+
+
+def test_one_quote_backs_one_position_only(events):
+ rows = _classify(
+ [
+ pos(events[0].event_key, "oppose", "a", EV_VAL),
+ pos(events[1].event_key, "oppose", "b", EV_VAL),
+ pos(events[2].event_key, "modify", "c", EV_G),
+ ],
+ events=events,
+ )
+ assert [(r.event_key, r.reason) for r in rows] == [
+ (events[0].event_key, "a"),
+ (events[2].event_key, "c"),
+ ]
+
+
+def test_reason_truncated(events):
+ rows = _classify(
+ [pos(events[0].event_key, "support", "x" * 500, EV_VAL)], events=events
+ )
assert len(rows[0].reason) == 200
+ assert raw_positions(None, events) is None
+ assert raw_positions({}, events) == []
+ assert (
+ raw_positions(
+ {"positions": [pos(events[0].event_key, "oppose", None, None)]}, events
+ )[0]["reason"]
+ == ""
+ )
+
+
+def test_position_rows_uses_given_extractor(events):
+ raw = [pos(events[0].event_key, "modify", "r", EV_VAL)]
+ (r,) = position_rows(raw, TEXT, family="F", item_key="K", docket="D", extractor="x")
+ assert (r.family, r.item_key, r.docket, r.extractor, r.stance) == (
+ "F",
+ "K",
+ "D",
+ "x",
+ "modify",
+ )
# ── stances_for (ledger) ─────────────────────────────────────────────
@@ -272,59 +493,68 @@ def _commenter(key, cid=None, org="individual"):
def _run(tmp_path, ask, commenters, texts, events, **kw):
log = RunLog(tmp_path / "stance.jsonl")
out = stances_for(
- commenters,
- texts,
- events,
- ask,
- log,
- family="HOME-VISITS",
- docket="D1",
- **kw,
+ commenters, texts, events, ask, log, family="HOME-VISITS", docket="D1", **kw
)
return out, log
-def test_stances_for_rows_from_ledger(tmp_path, events):
+def test_ledger_keeps_the_raw_reply_and_filters_run_on_replay(tmp_path, events):
ask = Ask(
[
pos(events[0].event_key, "support", "likes it", EV_VAL),
+ pos(events[0].event_key, "oppose", "dup", EV_VAL),
pos(events[1].event_key, "oppose", "bad", "not a quote from the text at all"),
pos("HOME-VISITS:2023:zzz", "support"),
]
) # fmt: skip
- (rows, counts), log = _run(tmp_path, ask, [_commenter("K1")], {"K1": TEXT}, events)
+ (rows, counts), _ = _run(tmp_path, ask, [_commenter("K1")], {"K1": TEXT}, events)
assert counts == {
"done": 1, "skipped": 0, "failed": 0, "failed_pending": 0,
"pending": 0, "no_text": 0,
} # fmt: skip
- assert [(r.event_key, r.reason, r.evidence) for r in rows] == [
- (events[0].event_key, "likes it", EV_VAL),
- (events[1].event_key, "bad", ""),
+ assert [(r.event_key, r.stance, r.reason) for r in rows] == [
+ (events[0].event_key, "support", "likes it")
]
assert rows[0].extractor == f"model:qwen-test@{STANCE_PROMPT_VERSION}"
- # a second run replays the ledger without asking again
- (again, _), _ = _run(tmp_path, ask, [_commenter("K1")], {"K1": TEXT}, events)
- assert again == rows and len(ask.calls) == 1
+ ledger = [
+ json.loads(line)
+ for line in (tmp_path / "stance.jsonl").read_text().splitlines()
+ ]
+ raw = ledger[0]["output"]["positions"]
+ assert (
+ len(raw) == 3 and raw[1]["stance"] == "oppose"
+ ) # everything valid, unfiltered
+ # replay with no model: same rows, the model is never asked
+ (again, c2), _ = _run(
+ tmp_path, None, [_commenter("K1")], {"K1": TEXT}, events, model="qwen-test"
+ )
+ assert again == rows and c2["done"] == 0 and c2["pending"] == 0
+ assert len(ask.calls) == 1
+
+
+def test_no_llm_without_a_ledger_is_pending(tmp_path, events):
+ (rows, counts), _ = _run(
+ tmp_path, None, [_commenter("K1")], {"K1": TEXT}, events, model="qwen-test"
+ )
+ assert rows == [] and counts["pending"] == 1
def test_stances_for_resume_limit_failed(tmp_path, events):
cs = [_commenter(k) for k in ("K1", "K2", "K3", "K4", "K5")]
texts = {k: TEXT for k in ("K1", "K2", "K3", "K4")} # K5 has no text
- ask = Ask([pos(events[3].event_key, "oppose")])
- (rows, counts), log = _run(tmp_path, ask, cs, texts, events, limit=2)
+ ask = Ask([pos(events[3].event_key, "oppose", "r", EV_VAL)])
+ (rows, counts), _ = _run(tmp_path, ask, cs, texts, events, limit=2)
assert counts["done"] == 2 and counts["pending"] == 2 and counts["no_text"] == 1
assert len(ask.calls) == 2 and len(rows) == 2
- # resume: the two done are skipped, the rest run
(rows, counts), _ = _run(tmp_path, ask, cs, texts, events)
assert counts["skipped"] == 2 and counts["done"] == 2 and counts["pending"] == 0
assert len(rows) == 4 and len(ask.calls) == 4
- # a failed reply is logged, stays pending-failed until --retry-failed
bad = Ask(None)
(rows, counts), _ = _run(tmp_path, bad, [_commenter("K9")], {"K9": TEXT}, events)
assert counts["failed"] == 1 and counts["pending"] == 1 and rows == []
(rows, counts), _ = _run(tmp_path, bad, [_commenter("K9")], {"K9": TEXT}, events)
assert counts["failed_pending"] == 1 and len(bad.calls) == 1
- good = Ask([pos(events[3].event_key, "support")])
+ good = Ask([pos(events[3].event_key, "support", "r", EV_VAL)])
(rows, counts), _ = _run(
tmp_path, good, [_commenter("K9")], {"K9": TEXT}, events, retry_failed=True
)
@@ -332,7 +562,7 @@ def test_stances_for_resume_limit_failed(tmp_path, events):
def test_stances_for_event_change_reruns(tmp_path, events):
- ask = Ask([pos(events[3].event_key, "support")])
+ ask = Ask([pos(events[3].event_key, "support", "r", EV_VAL)])
c, t = [_commenter("K1")], {"K1": TEXT}
_run(tmp_path, ask, c, t, events)
other = [*events[:3], EventChoice(events[3].event_key, "different text", 2023)]
@@ -342,9 +572,8 @@ def test_stances_for_event_change_reruns(tmp_path, events):
def test_stances_for_progress(tmp_path, events):
seen = []
- ask = Ask([])
_run(
- tmp_path, ask, [_commenter("K1")], {"K1": TEXT}, events,
+ tmp_path, Ask([]), [_commenter("K1")], {"K1": TEXT}, events,
progress=lambda u, s: seen.append((u, s)),
) # fmt: skip
assert seen == [("K1", "done")]
@@ -379,38 +608,60 @@ def _base(**kw):
return ReactionRow(**{**d, **kw})
+V, C, T, G, O = (
+ f"HOME-VISITS:2023:{s}"
+ for s in ("valuation", "coding", "telehealth", "G0318", "other")
+)
+ORDER = (V, C, T, G, O)
+
+
def test_rollup_counts_distinct_commenters():
- v, c, g, o = (
- f"HOME-VISITS:2023:{s}" for s in ("valuation", "coding", "G0318", "other")
- )
rows = [
# K1: valuation unclear, coding support -> first non-unclear = support
- S("K1", v, "unclear"), S("K1", c, "support"), S("K1", g, "oppose"),
- # K2: G0318 comes before other in event order, so modify wins
- S("K2", o, "oppose"), S("K2", o, "support"), S("K2", g, "modify"),
+ S("K1", V, "unclear"), S("K1", C, "support"), S("K1", G, "oppose"),
+ # K2: telehealth comes before G0318 and other in docket event order
+ S("K2", O, "oppose"), S("K2", O, "support"), S("K2", T, "modify"), S("K2", G, "oppose"),
# K3: all unclear
- S("K3", v, "unclear"), S("K3", o, "unclear"),
+ S("K3", V, "unclear"), S("K3", O, "unclear"),
# K4: same stance on several events counts once
- S("K4", v, "oppose"), S("K4", g, "oppose"),
+ S("K4", V, "oppose"), S("K4", G, "oppose"),
] # fmt: skip
- got = rollup(rows, base=_base())
+ got = rollup(rows, base=_base(), order=ORDER, n_commenters=52)
assert (
got.stance_support,
got.stance_modify,
got.stance_oppose,
got.stance_unclear,
) == (1, 1, 1, 1)
+ assert (got.n_classified, got.n_commenters) == (3, 52)
assert got.sample_json == '[{"k": 1}]' and got.n_items == 9 and got.period == "D1"
ev = json.loads(got.events_json)
- assert ev[v] == {"support": 0, "modify": 0, "oppose": 1, "unclear": 2}
- assert ev[c] == {"support": 1, "modify": 0, "oppose": 0, "unclear": 0}
- assert ev[g] == {"support": 0, "modify": 1, "oppose": 2, "unclear": 0}
- assert ev[o] == {"support": 0, "modify": 0, "oppose": 1, "unclear": 1}
+ assert list(ev) == [V, C, T, G, O] # docket_events order, not key-string order
+ assert ev[V] == {"support": 0, "modify": 0, "oppose": 1, "unclear": 2}
+ assert ev[C] == {"support": 1, "modify": 0, "oppose": 0, "unclear": 0}
+ assert ev[T] == {"support": 0, "modify": 1, "oppose": 0, "unclear": 0}
+ assert ev[G] == {"support": 0, "modify": 0, "oppose": 3, "unclear": 0}
+ assert ev[O] == {"support": 0, "modify": 0, "oppose": 1, "unclear": 1}
+
+
+def test_rollup_order_beats_key_order_and_unknown_events_sort_last():
+ # key-string order would put "G0318" before "telehealth"; docket order does not
+ rows = [
+ S("K1", G, "oppose"),
+ S("K1", T, "support"),
+ S("K1", "X:2023:zzz", "modify"),
+ ]
+ got = rollup(rows, base=_base(), order=ORDER)
+ assert got.stance_support == 1 and got.n_commenters is None
+ assert list(json.loads(got.events_json)) == [T, G, "X:2023:zzz"]
+ # with no order given, the event key decides
+ assert rollup(rows, base=_base()).stance_oppose == 1
def test_rollup_empty_is_zero():
- got = rollup([], base=_base(stance_support=3))
+ got = rollup([], base=_base(stance_support=3), n_commenters=4)
assert got.stance_support == 0 and got.events_json == "{}"
+ assert (got.n_classified, got.n_commenters) == (0, 4)
def test_rollup_ignores_cms_summaries(con):
@@ -420,6 +671,5 @@ def test_rollup_ignores_cms_summaries(con):
[CmsSummaryRow("HOME-VISITS", "R1", 2023, "final", 1, 2, "99342", "a majority of commenters", "support", "finalize_as_proposed", "regex")],
) # fmt: skip
rows = [S("K1", "HOME-VISITS:2023:other", "oppose")]
- assert rollup(rows, base=_base()) == rollup(rows, base=_base())
got = rollup(rows, base=_base())
assert (got.stance_support, got.stance_oppose) == (0, 1)