homelab/stack
1
0
Fork 0
Code Issues 212 Pull Requests 2 Actions Packages Projects 1 Releases 1 Wiki Activity

fix(pfs): stance-v2 — event texts state what CMS proposed, telehealth event, one verbatim span per position, raw ledger, reaction denominators

Browse Source
This commit is contained in:
kert
2026-11-12 13:26:43 -05:00
parent ba0fac3ad9
commit d4ec1ffc73
9 changed files with 819 additions and 213 deletions
Download Patch File Download Diff File Expand all files Collapse all files

46
src/cli/pfs.py
View File

@@ -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)

20
src/pfs/codetables.py
View File

@@ -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 = (

24
src/pfs/curated.py
View File

@@ -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 ())

2
src/pfs/curated.yaml
View File

@@ -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}}

451
src/pfs/stance.py
View File

@@ -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, ``<br/>`` to a
newline (regulations.gov stores inline comments with both)."""
return re.sub(r"<br\s*/?>", "\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,
)

41
tests/cli/test_pfs_cli.py
View File

@@ -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):

36
tests/pfs/test_curated.py
View File

@@ -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))

18
tests/pfs/test_reaction.py
View File

@@ -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:

394
tests/pfs/test_stance.py
View File

@@ -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&#39;s &quot;x&quot;<br/>next<br />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&#39;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)